Source code for fs.test

# coding: utf-8
"""Base class for tests.

All Filesystems should be able to pass these.

"""

from __future__ import absolute_import
from __future__ import unicode_literals

import collections
from datetime import datetime
import io
import itertools
import json
import math
import os
import time

import fs.copy
import fs.move
from fs import ResourceType, Seek
from fs import errors
from fs import walk
from fs import glob
from fs.opener import open_fs
from fs.subfs import ClosingSubFS, SubFS

import pytz
import six
from six import text_type


UNICODE_TEXT = """

UTF-8 encoded sample plain-text file
‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾‾

Markus Kuhn [ˈmaʳkʊs kuːn] <mkuhn@acm.org> — 1999-08-20


The ASCII compatible UTF-8 encoding of ISO 10646 and Unicode
plain-text files is defined in RFC 2279 and in ISO 10646-1 Annex R.


Using Unicode/UTF-8, you can write in emails and source code things such as

Mathematics and Sciences:

  ∮ E⋅da = Q,  n → ∞, ∑ f(i) = ∏ g(i), ∀x∈ℝ: ⌈x⌉ = −⌊−x⌋, α ∧ ¬β = ¬(¬α ∨ β),

  ℕ ⊆ ℕ₀ ⊂ ℤ ⊂ ℚ ⊂ ℝ ⊂ ℂ, ⊥ < a ≠ b ≡ c ≤ d ≪ ⊤ ⇒ (A ⇔ B),

  2H₂ + O₂ ⇌ 2H₂O, R = 4.7 kΩ, ⌀ 200 mm

Linguistics and dictionaries:

  ði ıntəˈnæʃənəl fəˈnɛtık əsoʊsiˈeıʃn
  Y [ˈʏpsilɔn], Yen [jɛn], Yoga [ˈjoːgɑ]

APL:

  ((V⍳V)=⍳⍴V)/V←,V    ⌷←⍳→⍴∆∇⊃‾⍎⍕⌈

Nicer typography in plain text files:

  ╔══════════════════════════════════════════╗
  ║                                          ║
  ║   • ‘single’ and “double” quotes         ║
  ║                                          ║
  ║   • Curly apostrophes: “We’ve been here” ║
  ║                                          ║
  ║   • Latin-1 apostrophe and accents: '´`  ║
  ║                                          ║
  ║   • ‚deutsche‘ „Anführungszeichen“       ║
  ║                                          ║
  ║   • †, ‡, ‰, •, 3–4, —, −5/+5, ™, …      ║
  ║                                          ║
  ║   • ASCII safety test: 1lI|, 0OD, 8B     ║
  ║                      ╭─────────╮         ║
  ║   • the euro symbol: │ 14.95 € │         ║
  ║                      ╰─────────╯         ║
  ╚══════════════════════════════════════════╝

Greek (in Polytonic):

  The Greek anthem:

  Σὲ γνωρίζω ἀπὸ τὴν κόψη
  τοῦ σπαθιοῦ τὴν τρομερή,
  σὲ γνωρίζω ἀπὸ τὴν ὄψη
  ποὺ μὲ βία μετράει τὴ γῆ.

  ᾿Απ᾿ τὰ κόκκαλα βγαλμένη
  τῶν ῾Ελλήνων τὰ ἱερά
  καὶ σὰν πρῶτα ἀνδρειωμένη
  χαῖρε, ὦ χαῖρε, ᾿Ελευθεριά!

  From a speech of Demosthenes in the 4th century BC:

  Οὐχὶ ταὐτὰ παρίσταταί μοι γιγνώσκειν, ὦ ἄνδρες ᾿Αθηναῖοι,
  ὅταν τ᾿ εἰς τὰ πράγματα ἀποβλέψω καὶ ὅταν πρὸς τοὺς
  λόγους οὓς ἀκούω· τοὺς μὲν γὰρ λόγους περὶ τοῦ
  τιμωρήσασθαι Φίλιππον ὁρῶ γιγνομένους, τὰ δὲ πράγματ᾿
  εἰς τοῦτο προήκοντα,  ὥσθ᾿ ὅπως μὴ πεισόμεθ᾿ αὐτοὶ
  πρότερον κακῶς σκέψασθαι δέον. οὐδέν οὖν ἄλλο μοι δοκοῦσιν
  οἱ τὰ τοιαῦτα λέγοντες ἢ τὴν ὑπόθεσιν, περὶ ἧς βουλεύεσθαι,
  οὐχὶ τὴν οὖσαν παριστάντες ὑμῖν ἁμαρτάνειν. ἐγὼ δέ, ὅτι μέν
  ποτ᾿ ἐξῆν τῇ πόλει καὶ τὰ αὑτῆς ἔχειν ἀσφαλῶς καὶ Φίλιππον
  τιμωρήσασθαι, καὶ μάλ᾿ ἀκριβῶς οἶδα· ἐπ᾿ ἐμοῦ γάρ, οὐ πάλαι
  γέγονεν ταῦτ᾿ ἀμφότερα· νῦν μέντοι πέπεισμαι τοῦθ᾿ ἱκανὸν
  προλαβεῖν ἡμῖν εἶναι τὴν πρώτην, ὅπως τοὺς συμμάχους
  σώσομεν. ἐὰν γὰρ τοῦτο βεβαίως ὑπάρξῃ, τότε καὶ περὶ τοῦ
  τίνα τιμωρήσεταί τις καὶ ὃν τρόπον ἐξέσται σκοπεῖν· πρὶν δὲ
  τὴν ἀρχὴν ὀρθῶς ὑποθέσθαι, μάταιον ἡγοῦμαι περὶ τῆς
  τελευτῆς ὁντινοῦν ποιεῖσθαι λόγον.

  Δημοσθένους, Γ´ ᾿Ολυνθιακὸς

Georgian:

  From a Unicode conference invitation:

  გთხოვთ ახლავე გაიაროთ რეგისტრაცია Unicode-ის მეათე საერთაშორისო
  კონფერენციაზე დასასწრებად, რომელიც გაიმართება 10-12 მარტს,
  ქ. მაინცში, გერმანიაში. კონფერენცია შეჰკრებს ერთად მსოფლიოს
  ექსპერტებს ისეთ დარგებში როგორიცაა ინტერნეტი და Unicode-ი,
  ინტერნაციონალიზაცია და ლოკალიზაცია, Unicode-ის გამოყენება
  ოპერაციულ სისტემებსა, და გამოყენებით პროგრამებში, შრიფტებში,
  ტექსტების დამუშავებასა და მრავალენოვან კომპიუტერულ სისტემებში.

Russian:

  From a Unicode conference invitation:

  Зарегистрируйтесь сейчас на Десятую Международную Конференцию по
  Unicode, которая состоится 10-12 марта 1997 года в Майнце в Германии.
  Конференция соберет широкий круг экспертов по  вопросам глобального
  Интернета и Unicode, локализации и интернационализации, воплощению и
  применению Unicode в различных операционных системах и программных
  приложениях, шрифтах, верстке и многоязычных компьютерных системах.

Thai (UCS Level 2):

  Excerpt from a poetry on The Romance of The Three Kingdoms (a Chinese
  classic 'San Gua'):

  [----------------------------|------------------------]
    ๏ แผ่นดินฮั่นเสื่อมโทรมแสนสังเวช  พระปกเกศกองบู๊กู้ขึ้นใหม่
  สิบสองกษัตริย์ก่อนหน้าแลถัดไป       สององค์ไซร้โง่เขลาเบาปัญญา
    ทรงนับถือขันทีเป็นที่พึ่ง           บ้านเมืองจึงวิปริตเป็นนักหนา
  โฮจิ๋นเรียกทัพทั่วหัวเมืองมา         หมายจะฆ่ามดชั่วตัวสำคัญ
    เหมือนขับไสไล่เสือจากเคหา      รับหมาป่าเข้ามาเลยอาสัญ
  ฝ่ายอ้องอุ้นยุแยกให้แตกกัน          ใช้สาวนั้นเป็นชนวนชื่นชวนใจ
    พลันลิฉุยกุยกีกลับก่อเหตุ          ช่างอาเพศจริงหนาฟ้าร้องไห้
  ต้องรบราฆ่าฟันจนบรรลัย           ฤๅหาใครค้ำชูกู้บรรลังก์ ฯ

  (The above is a two-column text. If combining characters are handled
  correctly, the lines of the second column should be aligned with the
  | character above.)

Ethiopian:

  Proverbs in the Amharic language:

  ሰማይ አይታረስ ንጉሥ አይከሰስ።
  ብላ ካለኝ እንደአባቴ በቆመጠኝ።
  ጌጥ ያለቤቱ ቁምጥና ነው።
  ደሀ በሕልሙ ቅቤ ባይጠጣ ንጣት በገደለው።
  የአፍ ወለምታ በቅቤ አይታሽም።
  አይጥ በበላ ዳዋ ተመታ።
  ሲተረጉሙ ይደረግሙ።
  ቀስ በቀስ፥ ዕንቁላል በእግሩ ይሄዳል።
  ድር ቢያብር አንበሳ ያስር።
  ሰው እንደቤቱ እንጅ እንደ ጉረቤቱ አይተዳደርም።
  እግዜር የከፈተውን ጉሮሮ ሳይዘጋው አይድርም።
  የጎረቤት ሌባ፥ ቢያዩት ይስቅ ባያዩት ያጠልቅ።
  ሥራ ከመፍታት ልጄን ላፋታት።
  ዓባይ ማደሪያ የለው፥ ግንድ ይዞ ይዞራል።
  የእስላም አገሩ መካ የአሞራ አገሩ ዋርካ።
  ተንጋሎ ቢተፉ ተመልሶ ባፉ።
  ወዳጅህ ማር ቢሆን ጨርስህ አትላሰው።
  እግርህን በፍራሽህ ልክ ዘርጋ።

Runes:

  ᚻᛖ ᚳᚹᚫᚦ ᚦᚫᛏ ᚻᛖ ᛒᚢᛞᛖ ᚩᚾ ᚦᚫᛗ ᛚᚪᚾᛞᛖ ᚾᚩᚱᚦᚹᛖᚪᚱᛞᚢᛗ ᚹᛁᚦ ᚦᚪ ᚹᛖᛥᚫ

  (Old English, which transcribed into Latin reads 'He cwaeth that he
  bude thaem lande northweardum with tha Westsae.' and means 'He said
  that he lived in the northern land near the Western Sea.')

Braille:

  ⡌⠁⠧⠑ ⠼⠁⠒  ⡍⠜⠇⠑⠹⠰⠎ ⡣⠕⠌

  ⡍⠜⠇⠑⠹ ⠺⠁⠎ ⠙⠑⠁⠙⠒ ⠞⠕ ⠃⠑⠛⠔ ⠺⠊⠹⠲ ⡹⠻⠑ ⠊⠎ ⠝⠕ ⠙⠳⠃⠞
  ⠱⠁⠞⠑⠧⠻ ⠁⠃⠳⠞ ⠹⠁⠞⠲ ⡹⠑ ⠗⠑⠛⠊⠌⠻ ⠕⠋ ⠙⠊⠎ ⠃⠥⠗⠊⠁⠇ ⠺⠁⠎
  ⠎⠊⠛⠝⠫ ⠃⠹ ⠹⠑ ⠊⠇⠻⠛⠹⠍⠁⠝⠂ ⠹⠑ ⠊⠇⠻⠅⠂ ⠹⠑ ⠥⠝⠙⠻⠞⠁⠅⠻⠂
  ⠁⠝⠙ ⠹⠑ ⠡⠊⠑⠋ ⠍⠳⠗⠝⠻⠲ ⡎⠊⠗⠕⠕⠛⠑ ⠎⠊⠛⠝⠫ ⠊⠞⠲ ⡁⠝⠙
  ⡎⠊⠗⠕⠕⠛⠑⠰⠎ ⠝⠁⠍⠑ ⠺⠁⠎ ⠛⠕⠕⠙ ⠥⠏⠕⠝ ⠰⡡⠁⠝⠛⠑⠂ ⠋⠕⠗ ⠁⠝⠹⠹⠔⠛ ⠙⠑
  ⠡⠕⠎⠑ ⠞⠕ ⠏⠥⠞ ⠙⠊⠎ ⠙⠁⠝⠙ ⠞⠕⠲

  ⡕⠇⠙ ⡍⠜⠇⠑⠹ ⠺⠁⠎ ⠁⠎ ⠙⠑⠁⠙ ⠁⠎ ⠁ ⠙⠕⠕⠗⠤⠝⠁⠊⠇⠲

  ⡍⠔⠙⠖ ⡊ ⠙⠕⠝⠰⠞ ⠍⠑⠁⠝ ⠞⠕ ⠎⠁⠹ ⠹⠁⠞ ⡊ ⠅⠝⠪⠂ ⠕⠋ ⠍⠹
  ⠪⠝ ⠅⠝⠪⠇⠫⠛⠑⠂ ⠱⠁⠞ ⠹⠻⠑ ⠊⠎ ⠏⠜⠞⠊⠊⠥⠇⠜⠇⠹ ⠙⠑⠁⠙ ⠁⠃⠳⠞
  ⠁ ⠙⠕⠕⠗⠤⠝⠁⠊⠇⠲ ⡊ ⠍⠊⠣⠞ ⠙⠁⠧⠑ ⠃⠑⠲ ⠔⠊⠇⠔⠫⠂ ⠍⠹⠎⠑⠇⠋⠂ ⠞⠕
  ⠗⠑⠛⠜⠙ ⠁ ⠊⠕⠋⠋⠔⠤⠝⠁⠊⠇ ⠁⠎ ⠹⠑ ⠙⠑⠁⠙⠑⠌ ⠏⠊⠑⠊⠑ ⠕⠋ ⠊⠗⠕⠝⠍⠕⠝⠛⠻⠹
  ⠔ ⠹⠑ ⠞⠗⠁⠙⠑⠲ ⡃⠥⠞ ⠹⠑ ⠺⠊⠎⠙⠕⠍ ⠕⠋ ⠳⠗ ⠁⠝⠊⠑⠌⠕⠗⠎
  ⠊⠎ ⠔ ⠹⠑ ⠎⠊⠍⠊⠇⠑⠆ ⠁⠝⠙ ⠍⠹ ⠥⠝⠙⠁⠇⠇⠪⠫ ⠙⠁⠝⠙⠎
  ⠩⠁⠇⠇ ⠝⠕⠞ ⠙⠊⠌⠥⠗⠃ ⠊⠞⠂ ⠕⠗ ⠹⠑ ⡊⠳⠝⠞⠗⠹⠰⠎ ⠙⠕⠝⠑ ⠋⠕⠗⠲ ⡹⠳
  ⠺⠊⠇⠇ ⠹⠻⠑⠋⠕⠗⠑ ⠏⠻⠍⠊⠞ ⠍⠑ ⠞⠕ ⠗⠑⠏⠑⠁⠞⠂ ⠑⠍⠏⠙⠁⠞⠊⠊⠁⠇⠇⠹⠂ ⠹⠁⠞
  ⡍⠜⠇⠑⠹ ⠺⠁⠎ ⠁⠎ ⠙⠑⠁⠙ ⠁⠎ ⠁ ⠙⠕⠕⠗⠤⠝⠁⠊⠇⠲

  (The first couple of paragraphs of "A Christmas Carol" by Dickens)

Compact font selection example text:

  ABCDEFGHIJKLMNOPQRSTUVWXYZ /0123456789
  abcdefghijklmnopqrstuvwxyz £©µÀÆÖÞßéöÿ
  –—‘“”„†•…‰™œŠŸž€ ΑΒΓΔΩαβγδω АБВГДабвгд
  ∀∂∈ℝ∧∪≡∞ ↑↗↨↻⇣ ┐┼╔╘░►☺♀ fi�⑀₂ἠḂӥẄɐː⍎אԱა

Greetings in various languages:

  Hello world, Καλημέρα κόσμε, コンニチハ

Box drawing alignment tests:                                          █

  ╔══╦══╗  ┌──┬──┐  ╭──┬──╮  ╭──┬──╮  ┏━━┳━━┓  ┎┒┏┑   ╷  ╻ ┏┯┓ ┌┰┐    ▊ ╱╲╱╲╳╳╳
  ║┌─╨─┐║  │╔═╧═╗│  │╒═╪═╕│  │╓─╁─╖│  ┃┌─╂─┐┃  ┗╃╄┙  ╶┼╴╺╋╸┠┼┨ ┝╋┥    ▋ ╲╱╲╱╳╳╳
  ║│╲ ╱│║  │║   ║│  ││ │ ││  │║ ┃ ║│  ┃│ ╿ │┃  ┍╅╆┓   ╵  ╹ ┗┷┛ └┸┘    ▌ ╱╲╱╲╳╳╳
  ╠╡ ╳ ╞╣  ├╢   ╟┤  ├┼─┼─┼┤  ├╫─╂─╫┤  ┣┿╾┼╼┿┫  ┕┛┖┚     ┌┄┄┐ ╎ ┏┅┅┓ ┋ ▍ ╲╱╲╱╳╳╳
  ║│╱ ╲│║  │║   ║│  ││ │ ││  │║ ┃ ║│  ┃│ ╽ │┃  ░░▒▒▓▓██ ┊  ┆ ╎ ╏  ┇ ┋ ▎
  ║└─╥─┘║  │╚═╤═╝│  │╘═╪═╛│  │╙─╀─╜│  ┃└─╂─┘┃  ░░▒▒▓▓██ ┊  ┆ ╎ ╏  ┇ ┋ ▏
  ╚══╩══╝  └──┴──┘  ╰──┴──╯  ╰──┴──╯  ┗━━┻━━┛           └╌╌┘ ╎ ┗╍╍┛ ┋  ▁▂▃▄▅▆▇█

"""


[docs]class FSTestCases(object): """Basic FS tests. """
[docs] def make_fs(self): """Return an FS instance. """ raise NotImplementedError("implement me")
[docs] def destroy_fs(self, fs): """Destroy a FS instance. Arguments: fs (FS): A filesystem instance previously opened by `~fs.test.FSTestCases.make_fs`. """ fs.close()
def setUp(self): self.fs = self.make_fs() def tearDown(self): self.destroy_fs(self.fs) del self.fs
[docs] def assert_exists(self, path): """Assert a path exists. Arguments: path (str): A path on the filesystem. """ self.assertTrue(self.fs.exists(path))
[docs] def assert_not_exists(self, path): """Assert a path does not exist. Arguments: path (str): A path on the filesystem. """ self.assertFalse(self.fs.exists(path))
[docs] def assert_isfile(self, path): """Assert a path is a file. Arguments: path (str): A path on the filesystem. """ self.assertTrue(self.fs.isfile(path))
[docs] def assert_isdir(self, path): """Assert a path is a directory. Arguments: path (str): A path on the filesystem. """ self.assertTrue(self.fs.isdir(path))
[docs] def assert_bytes(self, path, contents): """Assert a file contains the given bytes. Arguments: path (str): A path on the filesystem. contents (bytes): Bytes to compare. """ assert isinstance(contents, bytes) data = self.fs.readbytes(path) self.assertEqual(data, contents) self.assertIsInstance(data, bytes)
[docs] def assert_text(self, path, contents): """Assert a file contains the given text. Arguments: path (str): A path on the filesystem. contents (str): Text to compare. """ assert isinstance(contents, text_type) with self.fs.open(path, "rt") as f: data = f.read() self.assertEqual(data, contents) self.assertIsInstance(data, text_type)
def test_root_dir(self): with self.assertRaises(errors.FileExpected): self.fs.open("/") with self.assertRaises(errors.FileExpected): self.fs.openbin("/") def test_appendbytes(self): with self.assertRaises(TypeError): self.fs.appendbytes("foo", "bar") self.fs.appendbytes("foo", b"bar") self.assert_bytes("foo", b"bar") self.fs.appendbytes("foo", b"baz") self.assert_bytes("foo", b"barbaz") def test_appendtext(self): with self.assertRaises(TypeError): self.fs.appendtext("foo", b"bar") self.fs.appendtext("foo", "bar") self.assert_text("foo", "bar") self.fs.appendtext("foo", "baz") self.assert_text("foo", "barbaz") def test_basic(self): #  Check str and repr don't break repr(self.fs) self.assertIsInstance(six.text_type(self.fs), six.text_type) def test_getmeta(self): # Get the meta dict meta = self.fs.getmeta() # Check default namespace self.assertEqual(meta, self.fs.getmeta(namespace="standard")) # Must be a dict self.assertTrue(isinstance(meta, dict)) no_meta = self.fs.getmeta("__nosuchnamespace__") self.assertIsInstance(no_meta, dict) self.assertFalse(no_meta) def test_isfile(self): self.assertFalse(self.fs.isfile("foo.txt")) self.fs.create("foo.txt") self.assertTrue(self.fs.isfile("foo.txt")) self.fs.makedir("bar") self.assertFalse(self.fs.isfile("bar")) def test_isdir(self): self.assertFalse(self.fs.isdir("foo")) self.fs.create("bar") self.fs.makedir("foo") self.assertTrue(self.fs.isdir("foo")) self.assertFalse(self.fs.isdir("bar")) def test_islink(self): self.fs.touch("foo") self.assertFalse(self.fs.islink("foo")) with self.assertRaises(errors.ResourceNotFound): self.fs.islink("bar") def test_getsize(self): self.fs.writebytes("empty", b"") self.fs.writebytes("one", b"a") self.fs.writebytes("onethousand", ("b" * 1000).encode("ascii")) self.assertEqual(self.fs.getsize("empty"), 0) self.assertEqual(self.fs.getsize("one"), 1) self.assertEqual(self.fs.getsize("onethousand"), 1000) with self.assertRaises(errors.ResourceNotFound): self.fs.getsize("doesnotexist") def test_getsyspath(self): self.fs.create("foo") try: syspath = self.fs.getsyspath("foo") except errors.NoSysPath: self.assertFalse(self.fs.hassyspath("foo")) else: self.assertIsInstance(syspath, text_type) self.assertIsInstance(self.fs.getospath("foo"), bytes) self.assertTrue(self.fs.hassyspath("foo")) # Should not throw an error self.fs.hassyspath("a/b/c/foo/bar") def test_geturl(self): self.fs.create("foo") try: self.fs.geturl("foo") except errors.NoURL: self.assertFalse(self.fs.hasurl("foo")) else: self.assertTrue(self.fs.hasurl("foo")) # Should not throw an error self.fs.hasurl("a/b/c/foo/bar")
[docs] def test_geturl_purpose(self): """Check an unknown purpose raises a NoURL error. """ self.fs.create("foo") with self.assertRaises(errors.NoURL): self.fs.geturl("foo", purpose="__nosuchpurpose__")
[docs] def test_validatepath(self): """Check validatepath returns an absolute path. """ path = self.fs.validatepath("foo") self.assertEqual(path, "/foo")
def test_invalid_chars(self): # Test invalid path method. with self.assertRaises(errors.InvalidCharsInPath): self.fs.open("invalid\0file", "wb") with self.assertRaises(errors.InvalidCharsInPath): self.fs.validatepath("invalid\0file") def test_getinfo(self): # Test special case of root directory # Root directory has a name of '' root_info = self.fs.getinfo("/") self.assertEqual(root_info.name, "") self.assertTrue(root_info.is_dir) # Make a file of known size self.fs.writebytes("foo", b"bar") self.fs.makedir("dir") # Check basic namespace info = self.fs.getinfo("foo").raw self.assertIsInstance(info["basic"]["name"], text_type) self.assertEqual(info["basic"]["name"], "foo") self.assertFalse(info["basic"]["is_dir"]) # Check basic namespace dir info = self.fs.getinfo("dir").raw self.assertEqual(info["basic"]["name"], "dir") self.assertTrue(info["basic"]["is_dir"]) # Get the info info = self.fs.getinfo("foo", namespaces=["details"]).raw self.assertIsInstance(info, dict) self.assertEqual(info["details"]["size"], 3) self.assertEqual(info["details"]["type"], int(ResourceType.file)) # Test getdetails self.assertEqual(info, self.fs.getdetails("foo").raw) # Raw info should be serializable try: json.dumps(info) except: assert False, "info should be JSON serializable" # Non existant namespace is not an error no_info = self.fs.getinfo("foo", "__nosuchnamespace__").raw self.assertIsInstance(no_info, dict) self.assertEqual(no_info["basic"], {"name": "foo", "is_dir": False}) # Check a number of standard namespaces # FS objects may not support all these, but we can at least # invoke the code info = self.fs.getinfo("foo", namespaces=["access", "stat", "details"]) # Check that if the details namespace is present, times are # of valid types. if "details" in info.namespaces: details = info.raw["details"] self.assertIsInstance(details.get("accessed"), (type(None), int, float)) self.assertIsInstance(details.get("modified"), (type(None), int, float)) self.assertIsInstance(details.get("created"), (type(None), int, float)) self.assertIsInstance( details.get("metadata_changed"), (type(None), int, float) ) def test_exists(self): # Test exists method. # Check root directory always exists self.assertTrue(self.fs.exists("/")) self.assertTrue(self.fs.exists("")) # Check files don't exist self.assertFalse(self.fs.exists("foo")) self.assertFalse(self.fs.exists("foo/bar")) self.assertFalse(self.fs.exists("foo/bar/baz")) self.assertFalse(self.fs.exists("egg")) # make some files and directories self.fs.makedirs("foo/bar") self.fs.writebytes("foo/bar/baz", b"test") # Check files exists self.assertTrue(self.fs.exists("foo")) self.assertTrue(self.fs.exists("foo/bar")) self.assertTrue(self.fs.exists("foo/bar/baz")) self.assertFalse(self.fs.exists("egg")) self.assert_exists("foo") self.assert_exists("foo/bar") self.assert_exists("foo/bar/baz") self.assert_not_exists("egg") # Delete a file self.fs.remove("foo/bar/baz") # Check it no longer exists self.assert_not_exists("foo/bar/baz") self.assertFalse(self.fs.exists("foo/bar/baz")) self.assert_not_exists("foo/bar/baz") # Check root directory always exists self.assertTrue(self.fs.exists("/")) self.assertTrue(self.fs.exists("")) def test_listdir(self): # Check listing directory that doesn't exist with self.assertRaises(errors.ResourceNotFound): self.fs.listdir("foobar") # Check aliases for root self.assertEqual(self.fs.listdir("/"), []) self.assertEqual(self.fs.listdir("."), []) self.assertEqual(self.fs.listdir("./"), []) # Make a few objects self.fs.writebytes("foo", b"egg") self.fs.writebytes("bar", b"egg") self.fs.makedir("baz") # This should not be listed self.fs.writebytes("baz/egg", b"egg") # Check list works six.assertCountEqual(self, self.fs.listdir("/"), ["foo", "bar", "baz"]) six.assertCountEqual(self, self.fs.listdir("."), ["foo", "bar", "baz"]) six.assertCountEqual(self, self.fs.listdir("./"), ["foo", "bar", "baz"]) # Check paths are unicode strings for name in self.fs.listdir("/"): self.assertIsInstance(name, text_type) # Create a subdirectory self.fs.makedir("dir") # Should start empty self.assertEqual(self.fs.listdir("/dir"), []) # Write some files self.fs.writebytes("dir/foofoo", b"egg") self.fs.writebytes("dir/barbar", b"egg") # Check listing subdirectory six.assertCountEqual(self, self.fs.listdir("dir"), ["foofoo", "barbar"]) # Make sure they are unicode stringd for name in self.fs.listdir("dir"): self.assertIsInstance(name, text_type) self.fs.create("notadir") with self.assertRaises(errors.DirectoryExpected): self.fs.listdir("notadir") def test_move(self): # Make a file self.fs.writebytes("foo", b"egg") self.assert_isfile("foo") # Move it self.fs.move("foo", "bar") # Check it has gone from original location self.assert_not_exists("foo") # Check it exists in the new location, and contents match self.assert_exists("bar") self.assert_bytes("bar", b"egg") # Check moving to existing file fails self.fs.writebytes("foo2", b"eggegg") with self.assertRaises(errors.DestinationExists): self.fs.move("foo2", "bar") # Check move with overwrite=True self.fs.move("foo2", "bar", overwrite=True) self.assert_not_exists("foo2") # Check moving to a non-existant directory with self.assertRaises(errors.ResourceNotFound): self.fs.move("bar", "egg/bar") # Check moving an unexisting source with self.assertRaises(errors.ResourceNotFound): self.fs.move("egg", "spam") # Check moving between different directories self.fs.makedir("baz") self.fs.writebytes("baz/bazbaz", b"bazbaz") self.fs.makedir("baz2") self.fs.move("baz/bazbaz", "baz2/bazbaz") self.assert_not_exists("baz/bazbaz") self.assert_bytes("baz2/bazbaz", b"bazbaz") # Check moving a directory raises an error self.assert_isdir("baz2") self.assert_not_exists("yolk") with self.assertRaises(errors.FileExpected): self.fs.move("baz2", "yolk") def test_makedir(self): # Check edge case of root with self.assertRaises(errors.DirectoryExists): self.fs.makedir("/") # Making root is a null op with recreate slash_fs = self.fs.makedir("/", recreate=True) self.assertIsInstance(slash_fs, SubFS) self.assertEqual(self.fs.listdir("/"), []) self.assert_not_exists("foo") self.fs.makedir("foo") self.assert_isdir("foo") self.assertEqual(self.fs.gettype("foo"), ResourceType.directory) self.fs.writebytes("foo/bar.txt", b"egg") self.assert_bytes("foo/bar.txt", b"egg") # Directory exists with self.assertRaises(errors.DirectoryExists): self.fs.makedir("foo") # Parent directory doesn't exist with self.assertRaises(errors.ResourceNotFound): self.fs.makedir("/foo/bar/baz") self.fs.makedir("/foo/bar") self.fs.makedir("/foo/bar/baz") with self.assertRaises(errors.DirectoryExists): self.fs.makedir("foo/bar/baz") with self.assertRaises(errors.DirectoryExists): self.fs.makedir("foo/bar.txt") def test_makedirs(self): self.assertFalse(self.fs.exists("foo")) self.fs.makedirs("foo") self.assertEqual(self.fs.gettype("foo"), ResourceType.directory) self.fs.makedirs("foo/bar/baz") self.assertTrue(self.fs.isdir("foo/bar")) self.assertTrue(self.fs.isdir("foo/bar/baz")) with self.assertRaises(errors.DirectoryExists): self.fs.makedirs("foo/bar/baz") self.fs.makedirs("foo/bar/baz", recreate=True) self.fs.writebytes("foo.bin", b"test") with self.assertRaises(errors.DirectoryExpected): self.fs.makedirs("foo.bin/bar") with self.assertRaises(errors.DirectoryExpected): self.fs.makedirs("foo.bin/bar/baz/egg") def test_repeat_dir(self): # Catches bug with directories contain repeated names, # discovered in s3fs self.fs.makedirs("foo/foo/foo") self.assertEqual(self.fs.listdir(""), ["foo"]) self.assertEqual(self.fs.listdir("foo"), ["foo"]) self.assertEqual(self.fs.listdir("foo/foo"), ["foo"]) self.assertEqual(self.fs.listdir("foo/foo/foo"), []) scan = list(self.fs.scandir("foo")) self.assertEqual(len(scan), 1) self.assertEqual(scan[0].name, "foo") def test_open(self): # Open a file that doesn't exist with self.assertRaises(errors.ResourceNotFound): self.fs.open("doesnotexist", "r") self.fs.makedir("foo") # Create a new text file text = "Hello, World" with self.fs.open("foo/hello", "wt") as f: repr(f) self.assertIsInstance(f, io.IOBase) self.assertTrue(f.writable()) self.assertFalse(f.readable()) self.assertFalse(f.closed) f.write(text) self.assertTrue(f.closed) # Read it back with self.fs.open("foo/hello", "rt") as f: self.assertIsInstance(f, io.IOBase) self.assertTrue(f.readable()) self.assertFalse(f.writable()) self.assertFalse(f.closed) hello = f.read() self.assertTrue(f.closed) self.assertEqual(hello, text) self.assert_text("foo/hello", text) # Test overwrite text = "Goodbye, World" with self.fs.open("foo/hello", "wt") as f: f.write(text) self.assert_text("foo/hello", text) # Open from missing dir with self.assertRaises(errors.ResourceNotFound): self.fs.open("/foo/bar/test.txt") # Test fileno returns a file number, if supported by the file. with self.fs.open("foo/hello") as f: try: fn = f.fileno() except io.UnsupportedOperation: pass else: self.assertEqual(os.read(fn, 7), b"Goodbye") # Test text files are proper iterators over themselves lines = os.linesep.join(["Line 1", "Line 2", "Line 3"]) self.fs.writetext("iter.txt", lines) with self.fs.open("iter.txt") as f: for actual, expected in zip(f, lines.splitlines(1)): self.assertEqual(actual, expected) def test_openbin_rw(self): # Open a file that doesn't exist with self.assertRaises(errors.ResourceNotFound): self.fs.openbin("doesnotexist", "r") self.fs.makedir("foo") # Create a new text file text = b"Hello, World\n" with self.fs.openbin("foo/hello", "w") as f: repr(f) self.assertIsInstance(f, io.IOBase) self.assertTrue(f.writable()) self.assertFalse(f.readable()) self.assertEqual(len(text), f.write(text)) self.assertFalse(f.closed) self.assertTrue(f.closed) with self.assertRaises(errors.FileExists): with self.fs.openbin("foo/hello", "x") as f: pass # Read it back with self.fs.openbin("foo/hello", "r") as f: self.assertIsInstance(f, io.IOBase) self.assertTrue(f.readable()) self.assertFalse(f.writable()) hello = f.read() self.assertFalse(f.closed) self.assertTrue(f.closed) self.assertEqual(hello, text) self.assert_bytes("foo/hello", text) # Test overwrite text = b"Goodbye, World" with self.fs.openbin("foo/hello", "w") as f: self.assertEqual(len(text), f.write(text)) self.assert_bytes("foo/hello", text) # Test FileExpected raised with self.assertRaises(errors.FileExpected): self.fs.openbin("foo") # directory # Open from missing dir with self.assertRaises(errors.ResourceNotFound): self.fs.openbin("/foo/bar/test.txt") # Test fileno returns a file number, if supported by the file. with self.fs.openbin("foo/hello") as f: try: fn = f.fileno() except io.UnsupportedOperation: pass else: self.assertEqual(os.read(fn, 7), b"Goodbye") # Test binary files are proper iterators over themselves lines = b"\n".join([b"Line 1", b"Line 2", b"Line 3"]) self.fs.writebytes("iter.bin", lines) with self.fs.openbin("iter.bin") as f: for actual, expected in zip(f, lines.splitlines(1)): self.assertEqual(actual, expected) def test_open_files(self): # Test file-like objects work as expected. with self.fs.open("text", "w") as f: repr(f) text_type(f) self.assertIsInstance(f, io.IOBase) self.assertTrue(f.writable()) self.assertFalse(f.readable()) self.assertFalse(f.closed) self.assertEqual(f.tell(), 0) f.write("Hello\nWorld\n") self.assertEqual(f.tell(), 12) f.writelines(["foo\n", "bar\n", "baz\n"]) with self.assertRaises(IOError): f.read(1) self.assertTrue(f.closed) with self.fs.open("bin", "wb") as f: with self.assertRaises(IOError): f.read(1) with self.fs.open("text", "r") as f: repr(f) text_type(f) self.assertIsInstance(f, io.IOBase) self.assertFalse(f.writable()) self.assertTrue(f.readable()) self.assertFalse(f.closed) self.assertEqual( f.readlines(), ["Hello\n", "World\n", "foo\n", "bar\n", "baz\n"] ) with self.assertRaises(IOError): f.write("no") self.assertTrue(f.closed) with self.fs.open("text", "rb") as f: self.assertIsInstance(f, io.IOBase) self.assertFalse(f.writable()) self.assertTrue(f.readable()) self.assertFalse(f.closed) self.assertEqual(f.readlines(8), [b"Hello\n", b"World\n"]) with self.assertRaises(IOError): f.write(b"no") self.assertTrue(f.closed) with self.fs.open("text", "r") as f: self.assertEqual(list(f), ["Hello\n", "World\n", "foo\n", "bar\n", "baz\n"]) self.assertFalse(f.closed) self.assertTrue(f.closed) iter_lines = iter(self.fs.open("text")) self.assertEqual(next(iter_lines), "Hello\n") with self.fs.open("unicode", "w") as f: self.assertEqual(12, f.write("Héllo\nWörld\n")) with self.fs.open("text", "rb") as f: self.assertIsInstance(f, io.IOBase) self.assertFalse(f.writable()) self.assertTrue(f.readable()) self.assertTrue(f.seekable()) self.assertFalse(f.closed) self.assertEqual(f.read(1), b"H") self.assertEqual(3, f.seek(3, Seek.set)) self.assertEqual(f.read(1), b"l") self.assertEqual(6, f.seek(2, Seek.current)) self.assertEqual(f.read(1), b"W") self.assertEqual(22, f.seek(-2, Seek.end)) self.assertEqual(f.read(1), b"z") with self.assertRaises(ValueError): f.seek(10, 77) self.assertTrue(f.closed) with self.fs.open("text", "r+b") as f: self.assertIsInstance(f, io.IOBase) self.assertTrue(f.readable()) self.assertTrue(f.writable()) self.assertTrue(f.seekable()) self.assertFalse(f.closed) self.assertEqual(5, f.seek(5)) self.assertEqual(5, f.truncate()) self.assertEqual(0, f.seek(0)) self.assertEqual(f.read(), b"Hello") self.assertEqual(10, f.truncate(10)) self.assertEqual(5, f.tell()) self.assertEqual(0, f.seek(0)) print(repr(self.fs)) print(repr(f)) self.assertEqual(f.read(), b"Hello\0\0\0\0\0") self.assertEqual(4, f.seek(4)) f.write(b"O") self.assertEqual(4, f.seek(4)) self.assertEqual(f.read(1), b"O") self.assertTrue(f.closed) def test_openbin(self): # Write a binary file with self.fs.openbin("file.bin", "wb") as write_file: repr(write_file) text_type(write_file) self.assertIsInstance(write_file, io.IOBase) self.assertTrue(write_file.writable()) self.assertFalse(write_file.readable()) self.assertFalse(write_file.closed) self.assertEqual(3, write_file.write(b"\0\1\2")) self.assertTrue(write_file.closed) # Read a binary file with self.fs.openbin("file.bin", "rb") as read_file: repr(write_file) text_type(write_file) self.assertIsInstance(read_file, io.IOBase) self.assertTrue(read_file.readable()) self.assertFalse(read_file.writable()) self.assertFalse(read_file.closed) data = read_file.read() self.assertEqual(data, b"\0\1\2") self.assertTrue(read_file.closed) # Check disallow text mode with self.assertRaises(ValueError): with self.fs.openbin("file.bin", "rt") as read_file: pass # Check errors with self.assertRaises(errors.ResourceNotFound): self.fs.openbin("foo.bin") # Open from missing dir with self.assertRaises(errors.ResourceNotFound): self.fs.openbin("/foo/bar/test.txt") self.fs.makedir("foo") # Attempt to open a directory with self.assertRaises(errors.FileExpected): self.fs.openbin("/foo") # Attempt to write to a directory with self.assertRaises(errors.FileExpected): self.fs.openbin("/foo", "w") # Opening a file in a directory which doesn't exist with self.assertRaises(errors.ResourceNotFound): self.fs.openbin("/egg/bar") # Opening a file in a directory which doesn't exist with self.assertRaises(errors.ResourceNotFound): self.fs.openbin("/egg/bar", "w") # Opening with a invalid mode with self.assertRaises(ValueError): self.fs.openbin("foo.bin", "h") def test_open_exclusive(self): with self.fs.open("test_open_exclusive", "x") as f: f.write("bananas") with self.assertRaises(errors.FileExists): self.fs.open("test_open_exclusive", "x") def test_openbin_exclusive(self): with self.fs.openbin("test_openbin_exclusive", "x") as f: f.write(b"bananas") with self.assertRaises(errors.FileExists): self.fs.openbin("test_openbin_exclusive", "x") def test_opendir(self): # Make a simple directory structure self.fs.makedir("foo") self.fs.writebytes("foo/bar", b"barbar") self.fs.writebytes("foo/egg", b"eggegg") # Open a sub directory with self.fs.opendir("foo") as foo_fs: repr(foo_fs) text_type(foo_fs) six.assertCountEqual(self, foo_fs.listdir("/"), ["bar", "egg"]) self.assertTrue(foo_fs.isfile("bar")) self.assertTrue(foo_fs.isfile("egg")) self.assertEqual(foo_fs.readbytes("bar"), b"barbar") self.assertEqual(foo_fs.readbytes("egg"), b"eggegg") self.assertFalse(self.fs.isclosed()) # Attempt to open a non-existent directory with self.assertRaises(errors.ResourceNotFound): self.fs.opendir("egg") # Check error when doing opendir on a non dir with self.assertRaises(errors.DirectoryExpected): self.fs.opendir("foo/egg") # These should work, and will essentially return a 'clone' of sorts self.fs.opendir("") self.fs.opendir("/") # Check ClosingSubFS closes 'parent' with self.fs.opendir("foo", factory=ClosingSubFS) as foo_fs: six.assertCountEqual(self, foo_fs.listdir("/"), ["bar", "egg"]) self.assertTrue(foo_fs.isfile("bar")) self.assertTrue(foo_fs.isfile("egg")) self.assertEqual(foo_fs.readbytes("bar"), b"barbar") self.assertEqual(foo_fs.readbytes("egg"), b"eggegg") self.assertTrue(self.fs.isclosed()) def test_remove(self): self.fs.writebytes("foo1", b"test1") self.fs.writebytes("foo2", b"test2") self.fs.writebytes("foo3", b"test3") self.assert_isfile("foo1") self.assert_isfile("foo2") self.assert_isfile("foo3") self.fs.remove("foo2") self.assert_isfile("foo1") self.assert_not_exists("foo2") self.assert_isfile("foo3") with self.assertRaises(errors.ResourceNotFound): self.fs.remove("bar") self.fs.makedir("dir") with self.assertRaises(errors.FileExpected): self.fs.remove("dir") self.fs.makedirs("foo/bar/baz/") error_msg = "resource 'foo/bar/egg/test.txt' not found" with self.assertRaisesRegexp(errors.ResourceNotFound, error_msg): self.fs.remove("foo/bar/egg/test.txt") def test_removedir(self): # Test removing root with self.assertRaises(errors.RemoveRootError): self.fs.removedir("/") self.fs.makedirs("foo/bar/baz") self.assertTrue(self.fs.exists("foo/bar/baz")) self.fs.removedir("foo/bar/baz") self.assertFalse(self.fs.exists("foo/bar/baz")) self.assertTrue(self.fs.isdir("foo/bar")) with self.assertRaises(errors.ResourceNotFound): self.fs.removedir("nodir") # Test force removal self.fs.makedirs("foo/bar/baz") self.fs.writebytes("foo/egg", b"test") with self.assertRaises(errors.DirectoryExpected): self.fs.removedir("foo/egg") with self.assertRaises(errors.DirectoryNotEmpty): self.fs.removedir("foo/bar") def test_removetree(self): self.fs.makedirs("foo/bar/baz") self.fs.makedirs("foo/egg") self.fs.makedirs("foo/a/b/c/d/e") self.fs.create("foo/egg.txt") self.fs.create("foo/bar/egg.bin") self.fs.create("foo/bar/baz/egg.txt") self.fs.create("foo/a/b/c/1.txt") self.fs.create("foo/a/b/c/2.txt") self.fs.create("foo/a/b/c/3.txt") self.assert_exists("foo/egg.txt") self.assert_exists("foo/bar/egg.bin") self.fs.removetree("foo") self.assert_not_exists("foo") def test_setinfo(self): self.fs.create("birthday.txt") now = math.floor(time.time()) change_info = {"details": {"accessed": now + 60, "modified": now + 60 * 60}} self.fs.setinfo("birthday.txt", change_info) new_info = self.fs.getinfo("birthday.txt", namespaces=["details"]).raw if "accessed" in new_info.get("_write", []): self.assertEqual(new_info["details"]["accessed"], now + 60) if "modified" in new_info.get("_write", []): self.assertEqual(new_info["details"]["modified"], now + 60 * 60) with self.assertRaises(errors.ResourceNotFound): self.fs.setinfo("nothing", {}) def test_settimes(self): self.fs.create("birthday.txt") self.fs.settimes("birthday.txt", accessed=datetime(2016, 7, 5)) info = self.fs.getinfo("birthday.txt", namespaces=["details"]) writeable = info.get("details", "_write", []) if "accessed" in writeable: self.assertEqual(info.accessed, datetime(2016, 7, 5, tzinfo=pytz.UTC)) if "modified" in writeable: self.assertEqual(info.modified, datetime(2016, 7, 5, tzinfo=pytz.UTC)) def test_touch(self): self.fs.touch("new.txt") self.assert_isfile("new.txt") self.fs.settimes("new.txt", datetime(2016, 7, 5)) info = self.fs.getinfo("new.txt", namespaces=["details"]) if info.is_writeable("details", "accessed"): self.assertEqual(info.accessed, datetime(2016, 7, 5, tzinfo=pytz.UTC)) now = time.time() self.fs.touch("new.txt") accessed = self.fs.getinfo("new.txt", namespaces=["details"]).raw[ "details" ]["accessed"] self.assertTrue(accessed - now < 5) def test_close(self): self.assertFalse(self.fs.isclosed()) self.fs.close() self.assertTrue(self.fs.isclosed()) # Check second close call is a no-op self.fs.close() self.assertTrue(self.fs.isclosed()) # Check further operations raise a FilesystemClosed exception with self.assertRaises(errors.FilesystemClosed): self.fs.openbin("test.bin") def test_copy(self): # Test copy to new path self.fs.writebytes("foo", b"test") self.fs.copy("foo", "bar") self.assert_bytes("bar", b"test") # Test copy over existing path self.fs.writebytes("baz", b"truncateme") self.fs.copy("foo", "baz", overwrite=True) self.assert_bytes("foo", b"test") # Test copying a file to a destination that exists with self.assertRaises(errors.DestinationExists): self.fs.copy("baz", "foo") # Test copying to a directory that doesn't exist with self.assertRaises(errors.ResourceNotFound): self.fs.copy("baz", "a/b/c/baz") # Test copying a source that doesn't exist with self.assertRaises(errors.ResourceNotFound): self.fs.copy("egg", "spam") # Test copying a directory self.fs.makedir("dir") with self.assertRaises(errors.FileExpected): self.fs.copy("dir", "folder") def _test_upload(self, workers): """Test fs.copy with varying number of worker threads.""" data1 = b"foo" * 256 * 1024 data2 = b"bar" * 2 * 256 * 1024 data3 = b"baz" * 3 * 256 * 1024 data4 = b"egg" * 7 * 256 * 1024 with open_fs("temp://") as src_fs: src_fs.writebytes("foo", data1) src_fs.writebytes("bar", data2) src_fs.makedir("dir1").writebytes("baz", data3) src_fs.makedirs("dir2/dir3").writebytes("egg", data4) dst_fs = self.fs fs.copy.copy_fs(src_fs, dst_fs, workers=workers) self.assertEqual(dst_fs.readbytes("foo"), data1) self.assertEqual(dst_fs.readbytes("bar"), data2) self.assertEqual(dst_fs.readbytes("dir1/baz"), data3) self.assertEqual(dst_fs.readbytes("dir2/dir3/egg"), data4) def test_upload_0(self): self._test_upload(0) def test_upload_1(self): self._test_upload(1) def test_upload_2(self): self._test_upload(2) def test_upload_4(self): self._test_upload(4) def _test_download(self, workers): """Test fs.copy with varying number of worker threads.""" data1 = b"foo" * 256 * 1024 data2 = b"bar" * 2 * 256 * 1024 data3 = b"baz" * 3 * 256 * 1024 data4 = b"egg" * 7 * 256 * 1024 src_fs = self.fs with open_fs("temp://") as dst_fs: src_fs.writebytes("foo", data1) src_fs.writebytes("bar", data2) src_fs.makedir("dir1").writebytes("baz", data3) src_fs.makedirs("dir2/dir3").writebytes("egg", data4) fs.copy.copy_fs(src_fs, dst_fs, workers=workers) self.assertEqual(dst_fs.readbytes("foo"), data1) self.assertEqual(dst_fs.readbytes("bar"), data2) self.assertEqual(dst_fs.readbytes("dir1/baz"), data3) self.assertEqual(dst_fs.readbytes("dir2/dir3/egg"), data4) def test_download_0(self): self._test_download(0) def test_download_1(self): self._test_download(1) def test_download_2(self): self._test_download(2) def test_download_4(self): self._test_download(4) def test_create(self): # Test create new file self.assertFalse(self.fs.exists("foo")) self.fs.create("foo") self.assertTrue(self.fs.exists("foo")) self.assertEqual(self.fs.gettype("foo"), ResourceType.file) self.assertEqual(self.fs.getsize("foo"), 0) # Test wipe existing file self.fs.writebytes("foo", b"bar") self.assertEqual(self.fs.getsize("foo"), 3) self.fs.create("foo", wipe=True) self.assertEqual(self.fs.getsize("foo"), 0) # Test create with existing file, and not wipe self.fs.writebytes("foo", b"bar") self.assertEqual(self.fs.getsize("foo"), 3) self.fs.create("foo", wipe=False) self.assertEqual(self.fs.getsize("foo"), 3) def test_desc(self): # Describe a file self.fs.create("foo") description = self.fs.desc("foo") self.assertIsInstance(description, text_type) # Describe a dir self.fs.makedir("dir") self.fs.desc("dir") # Special cases that may hide bugs self.fs.desc("/") self.fs.desc("") with self.assertRaises(errors.ResourceNotFound): self.fs.desc("bar") def test_scandir(self): # Check exception for scanning dir that doesn't exist with self.assertRaises(errors.ResourceNotFound): for info in self.fs.scandir("/foobar"): pass # Check scandir returns an iterable iter_scandir = self.fs.scandir("/") self.assertTrue(isinstance(iter_scandir, collections.Iterable)) self.assertEqual(list(iter_scandir), []) # Check scanning self.fs.create("foo") # Can't scandir on a file with self.assertRaises(errors.DirectoryExpected): list(self.fs.scandir("foo")) self.fs.create("bar") self.fs.makedir("dir") iter_scandir = self.fs.scandir("/") self.assertTrue(isinstance(iter_scandir, collections.Iterable)) scandir = sorted( [r.raw for r in iter_scandir], key=lambda info: info["basic"]["name"] ) # Filesystems may send us more than we ask for # We just want to test the 'basic' namespace scandir = [{"basic": i["basic"]} for i in scandir] self.assertEqual( scandir, [ {"basic": {"name": "bar", "is_dir": False}}, {"basic": {"name": "dir", "is_dir": True}}, {"basic": {"name": "foo", "is_dir": False}}, ], ) # Hard to test optional namespaces, but at least run the code list( self.fs.scandir( "/", namespaces=["details", "link", "stat", "lstat", "access"] ) ) # Test paging page1 = list(self.fs.scandir("/", page=(None, 2))) self.assertEqual(len(page1), 2) page2 = list(self.fs.scandir("/", page=(2, 4))) self.assertEqual(len(page2), 1) page3 = list(self.fs.scandir("/", page=(4, 6))) self.assertEqual(len(page3), 0) paged = set(r.name for r in itertools.chain(page1, page2)) self.assertEqual(paged, {"foo", "bar", "dir"}) def test_filterdir(self): self.assertEqual(list(self.fs.filterdir("/", files=["*.py"])), []) self.fs.makedir("bar") self.fs.create("foo.txt") self.fs.create("foo.py") self.fs.create("foo.pyc") page1 = list(self.fs.filterdir("/", page=(None, 2))) page2 = list(self.fs.filterdir("/", page=(2, 4))) page3 = list(self.fs.filterdir("/", page=(4, 6))) self.assertEqual(len(page1), 2) self.assertEqual(len(page2), 2) self.assertEqual(len(page3), 0) names = [info.name for info in itertools.chain(page1, page2, page3)] self.assertEqual(set(names), {"foo.txt", "foo.py", "foo.pyc", "bar"}) # Check filtering by wildcard dir_list = [info.name for info in self.fs.filterdir("/", files=["*.py"])] self.assertEqual(set(dir_list), {"bar", "foo.py"}) # Check filtering by miltiple wildcard dir_list = [ info.name for info in self.fs.filterdir("/", files=["*.py", "*.pyc"]) ] self.assertEqual(set(dir_list), {"bar", "foo.py", "foo.pyc"}) # Check excluding dirs dir_list = [ info.name for info in self.fs.filterdir( "/", exclude_dirs=["*"], files=["*.py", "*.pyc"] ) ] self.assertEqual(set(dir_list), {"foo.py", "foo.pyc"}) # Check excluding files dir_list = [info.name for info in self.fs.filterdir("/", exclude_files=["*"])] self.assertEqual(set(dir_list), {"bar"}) # Check wildcards must be a list with self.assertRaises(TypeError): dir_list = [info.name for info in self.fs.filterdir("/", files="*.py")] self.fs.makedir("baz") dir_list = [ info.name for info in self.fs.filterdir("/", exclude_files=["*"], dirs=["??z"]) ] self.assertEqual(set(dir_list), {"baz"}) with self.assertRaises(TypeError): dir_list = [ info.name for info in self.fs.filterdir("/", exclude_files=["*"], dirs="*.py") ] def test_readbytes(self): # Test readbytes method. all_bytes = b"".join(six.int2byte(n) for n in range(256)) with self.fs.open("foo", "wb") as f: f.write(all_bytes) self.assertEqual(self.fs.readbytes("foo"), all_bytes) _all_bytes = self.fs.readbytes("foo") self.assertIsInstance(_all_bytes, bytes) self.assertEqual(_all_bytes, all_bytes) with self.assertRaises(errors.ResourceNotFound): self.fs.readbytes("foo/bar") self.fs.makedir("baz") with self.assertRaises(errors.FileExpected): self.fs.readbytes("baz") def test_download(self): test_bytes = b"Hello, World" self.fs.writebytes("hello.bin", test_bytes) write_file = io.BytesIO() self.fs.download("hello.bin", write_file) self.assertEqual(write_file.getvalue(), test_bytes) with self.assertRaises(errors.ResourceNotFound): self.fs.download("foo.bin", write_file) def test_download_chunk_size(self): test_bytes = b"Hello, World" * 100 self.fs.writebytes("hello.bin", test_bytes) write_file = io.BytesIO() self.fs.download("hello.bin", write_file, chunk_size=8) self.assertEqual(write_file.getvalue(), test_bytes) def test_isempty(self): self.assertTrue(self.fs.isempty("/")) self.fs.makedir("foo") self.assertFalse(self.fs.isempty("/")) self.assertTrue(self.fs.isempty("/foo")) self.fs.create("foo/bar.txt") self.assertFalse(self.fs.isempty("/foo")) self.fs.remove("foo/bar.txt") self.assertTrue(self.fs.isempty("/foo")) def test_writebytes(self): all_bytes = b"".join(six.int2byte(n) for n in range(256)) self.fs.writebytes("foo", all_bytes) with self.fs.open("foo", "rb") as f: _bytes = f.read() self.assertIsInstance(_bytes, bytes) self.assertEqual(_bytes, all_bytes) self.assert_bytes("foo", all_bytes) with self.assertRaises(TypeError): self.fs.writebytes("notbytes", "unicode") def test_readtext(self): self.fs.makedir("foo") with self.fs.open("foo/unicode.txt", "wt") as f: f.write(UNICODE_TEXT) text = self.fs.readtext("foo/unicode.txt") self.assertIsInstance(text, text_type) self.assertEqual(text, UNICODE_TEXT) self.assert_text("foo/unicode.txt", UNICODE_TEXT) def test_writetext(self): # Test writetext method. self.fs.writetext("foo", "bar") with self.fs.open("foo", "rt") as f: foo = f.read() self.assertEqual(foo, "bar") self.assertIsInstance(foo, text_type) with self.assertRaises(TypeError): self.fs.writetext("nottext", b"bytes") def test_writefile(self): bytes_file = io.BytesIO(b"bar") self.fs.writefile("foo", bytes_file) with self.fs.open("foo", "rb") as f: data = f.read() self.assertEqual(data, b"bar") def test_upload(self): bytes_file = io.BytesIO(b"bar") self.fs.upload("foo", bytes_file) with self.fs.open("foo", "rb") as f: data = f.read() self.assertEqual(data, b"bar") def test_upload_chunk_size(self): test_data = b"bar" * 128 bytes_file = io.BytesIO(test_data) self.fs.upload("foo", bytes_file, chunk_size=8) with self.fs.open("foo", "rb") as f: data = f.read() self.assertEqual(data, test_data) def test_bin_files(self): # Check binary files. with self.fs.openbin("foo1", "wb") as f: text_type(f) repr(f) f.write(b"a") f.write(b"b") f.write(b"c") self.assert_bytes("foo1", b"abc") # Test writelines with self.fs.openbin("foo2", "wb") as f: f.writelines([b"hello\n", b"world"]) self.assert_bytes("foo2", b"hello\nworld") # Test readline with self.fs.openbin("foo2") as f: self.assertEqual(f.readline(), b"hello\n") self.assertEqual(f.readline(), b"world") # Test readlines with self.fs.openbin("foo2") as f: lines = f.readlines() self.assertEqual(lines, [b"hello\n", b"world"]) with self.fs.openbin("foo2") as f: lines = list(f) self.assertEqual(lines, [b"hello\n", b"world"]) with self.fs.openbin("foo2") as f: lines = [] for line in f: lines.append(line) self.assertEqual(lines, [b"hello\n", b"world"]) with self.fs.openbin("foo2") as f: print(repr(f)) self.assertEqual(next(f), b"hello\n") # Test truncate with self.fs.open("foo2", "r+b") as f: f.truncate(3) self.assertEqual(self.fs.getsize("foo2"), 3) self.assert_bytes("foo2", b"hel") def test_files(self): # Test multiple writes with self.fs.open("foo1", "wt") as f: text_type(f) repr(f) f.write("a") f.write("b") f.write("c") self.assert_text("foo1", "abc") # Test writelines with self.fs.open("foo2", "wt") as f: f.writelines(["hello\n", "world"]) self.assert_text("foo2", "hello\nworld") # Test readline with self.fs.open("foo2") as f: self.assertEqual(f.readline(), "hello\n") self.assertEqual(f.readline(), "world") # Test readlines with self.fs.open("foo2") as f: lines = f.readlines() self.assertEqual(lines, ["hello\n", "world"]) with self.fs.open("foo2") as f: lines = list(f) self.assertEqual(lines, ["hello\n", "world"]) with self.fs.open("foo2") as f: lines = [] for line in f: lines.append(line) self.assertEqual(lines, ["hello\n", "world"]) # Test truncate with self.fs.open("foo2", "r+") as f: f.truncate(3) self.assertEqual(self.fs.getsize("foo2"), 3) self.assert_text("foo2", "hel") with self.fs.open("foo2", "ab") as f: f.write(b"p") self.assert_bytes("foo2", b"help") # Test __del__ doesn't throw traceback f = self.fs.open("foo2", "r") del f with self.assertRaises(IOError): with self.fs.open("foo2", "r") as f: f.write("no!") with self.assertRaises(IOError): with self.fs.open("newfoo", "w") as f: f.read(2) def test_copy_file(self): # Test fs.copy.copy_file bytes_test = b"Hello, World" self.fs.writebytes("foo.txt", bytes_test) fs.copy.copy_file(self.fs, "foo.txt", self.fs, "bar.txt") self.assert_bytes("bar.txt", bytes_test) mem_fs = open_fs("mem://") fs.copy.copy_file(self.fs, "foo.txt", mem_fs, "bar.txt") self.assertEqual(mem_fs.readbytes("bar.txt"), bytes_test) def test_copy_structure(self): mem_fs = open_fs("mem://") self.fs.makedirs("foo/bar/baz") self.fs.makedir("egg") fs.copy.copy_structure(self.fs, mem_fs) expected = {"/egg", "/foo", "/foo/bar", "/foo/bar/baz"} self.assertEqual(set(walk.walk_dirs(mem_fs)), expected) def _test_copy_dir(self, protocol): # Test copy.copy_dir. # Test copying to a another fs other_fs = open_fs(protocol) self.fs.makedirs("foo/bar/baz") self.fs.makedir("egg") self.fs.writetext("top.txt", "Hello, World") self.fs.writetext("/foo/bar/baz/test.txt", "Goodbye, World") fs.copy.copy_dir(self.fs, "/", other_fs, "/") expected = {"/egg", "/foo", "/foo/bar", "/foo/bar/baz"} self.assertEqual(set(walk.walk_dirs(other_fs)), expected) self.assert_text("top.txt", "Hello, World") self.assert_text("/foo/bar/baz/test.txt", "Goodbye, World") # Test copying a sub dir other_fs = open_fs("mem://") fs.copy.copy_dir(self.fs, "/foo", other_fs, "/") self.assertEqual(list(walk.walk_files(other_fs)), ["/bar/baz/test.txt"]) print("BEFORE") self.fs.tree() other_fs.tree() fs.copy.copy_dir(self.fs, "/foo", other_fs, "/egg") print("FS") self.fs.tree() print("OTHER") other_fs.tree() self.assertEqual( list(walk.walk_files(other_fs)), ["/bar/baz/test.txt", "/egg/bar/baz/test.txt"], ) def _test_copy_dir_write(self, protocol): # Test copying to this filesystem from another. other_fs = open_fs(protocol) other_fs.makedirs("foo/bar/baz") other_fs.makedir("egg") other_fs.writetext("top.txt", "Hello, World") other_fs.writetext("/foo/bar/baz/test.txt", "Goodbye, World") fs.copy.copy_dir(other_fs, "/", self.fs, "/") expected = {"/egg", "/foo", "/foo/bar", "/foo/bar/baz"} self.assertEqual(set(walk.walk_dirs(self.fs)), expected) self.assert_text("top.txt", "Hello, World") self.assert_text("/foo/bar/baz/test.txt", "Goodbye, World") def test_copy_dir_mem(self): # Test copy_dir with a mem fs. self._test_copy_dir("mem://") self._test_copy_dir_write("mem://") def test_copy_dir_temp(self): # Test copy_dir with a temp fs. self._test_copy_dir("temp://") self._test_copy_dir_write("temp://") def _test_move_dir_write(self, protocol): # Test moving to this filesystem from another. other_fs = open_fs(protocol) other_fs.makedirs("foo/bar/baz") other_fs.makedir("egg") other_fs.writetext("top.txt", "Hello, World") other_fs.writetext("/foo/bar/baz/test.txt", "Goodbye, World") fs.move.move_dir(other_fs, "/", self.fs, "/") expected = {"/egg", "/foo", "/foo/bar", "/foo/bar/baz"} self.assertEqual(other_fs.listdir("/"), []) self.assertEqual(set(walk.walk_dirs(self.fs)), expected) self.assert_text("top.txt", "Hello, World") self.assert_text("/foo/bar/baz/test.txt", "Goodbye, World") def test_move_dir_mem(self): self._test_move_dir_write("mem://") def test_move_dir_temp(self): self._test_move_dir_write("temp://") def test_move_same_fs(self): self.fs.makedirs("foo/bar/baz") self.fs.makedir("egg") self.fs.writetext("top.txt", "Hello, World") self.fs.writetext("/foo/bar/baz/test.txt", "Goodbye, World") fs.move.move_dir(self.fs, "foo", self.fs, "foo2") expected = {"/egg", "/foo2", "/foo2/bar", "/foo2/bar/baz"} self.assertEqual(set(walk.walk_dirs(self.fs)), expected) self.assert_text("top.txt", "Hello, World") self.assert_text("/foo2/bar/baz/test.txt", "Goodbye, World") def test_move_file_same_fs(self): text = "Hello, World" self.fs.makedir("foo").writetext("test.txt", text) self.assert_text("foo/test.txt", text) fs.move.move_file(self.fs, "foo/test.txt", self.fs, "foo/test2.txt") self.assert_not_exists("foo/test.txt") self.assert_text("foo/test2.txt", text) def _test_move_file(self, protocol): other_fs = open_fs(protocol) text = "Hello, World" self.fs.makedir("foo").writetext("test.txt", text) self.assert_text("foo/test.txt", text) with self.assertRaises(errors.ResourceNotFound): fs.move.move_file(self.fs, "foo/test.txt", other_fs, "foo/test2.txt") other_fs.makedir("foo") fs.move.move_file(self.fs, "foo/test.txt", other_fs, "foo/test2.txt") self.assertEqual(other_fs.readtext("foo/test2.txt"), text) def test_move_file_mem(self): self._test_move_file("mem://") def test_move_file_temp(self): self._test_move_file("temp://") def test_copydir(self): self.fs.makedirs("foo/bar/baz/egg") self.fs.writetext("foo/bar/foofoo.txt", "Hello") self.fs.makedir("foo2") self.fs.copydir("foo/bar", "foo2") self.assert_text("foo2/foofoo.txt", "Hello") self.assert_isdir("foo2/baz/egg") self.assert_text("foo/bar/foofoo.txt", "Hello") self.assert_isdir("foo/bar/baz/egg") with self.assertRaises(errors.ResourceNotFound): self.fs.copydir("foo", "foofoo") with self.assertRaises(errors.ResourceNotFound): self.fs.copydir("spam", "egg", create=True) with self.assertRaises(errors.DirectoryExpected): self.fs.copydir("foo2/foofoo.txt", "foofoo.txt", create=True) def test_movedir(self): self.fs.makedirs("foo/bar/baz/egg") self.fs.writetext("foo/bar/foofoo.txt", "Hello") self.fs.makedir("foo2") self.fs.movedir("foo/bar", "foo2") self.assert_text("foo2/foofoo.txt", "Hello") self.assert_isdir("foo2/baz/egg") self.assert_not_exists("foo/bar") self.assert_not_exists("foo/bar/foofoo.txt") self.assert_not_exists("foo/bar/baz/egg") # Check moving to an unexisting directory with self.assertRaises(errors.ResourceNotFound): self.fs.movedir("foo", "foofoo") # Check moving an unexisting directory with self.assertRaises(errors.ResourceNotFound): self.fs.movedir("spam", "egg", create=True) # Check moving a file with self.assertRaises(errors.DirectoryExpected): self.fs.movedir("foo2/foofoo.txt", "foo2/baz/egg") def test_match(self): self.assertTrue(self.fs.match(["*.py"], "foo.py")) self.assertEqual( self.fs.match(["*.py"], "FOO.PY"), self.fs.getmeta().get("case_insensitive", False), ) def test_tree(self): self.fs.makedirs("foo/bar") self.fs.create("test.txt") write_tree = io.StringIO() self.fs.tree(file=write_tree) written = write_tree.getvalue() expected = "|-- foo\n| `-- bar\n`-- test.txt\n" self.assertEqual(expected, written) def test_unicode_path(self): if not self.fs.getmeta().get("unicode_paths", False): self.skipTest("the filesystem does not support unicode paths.") self.fs.makedir("földér") self.fs.writetext("☭.txt", "Smells like communism.") self.fs.writebytes("földér/☣.txt", b"Smells like an old syringe.") self.assert_isdir("földér") self.assertEqual(["☣.txt"], self.fs.listdir("földér")) self.assertEqual("☣.txt", self.fs.getinfo("földér/☣.txt").name) self.assert_text("☭.txt", "Smells like communism.") self.assert_bytes("földér/☣.txt", b"Smells like an old syringe.") if self.fs.hassyspath("földér/☣.txt"): self.assertTrue(os.path.exists(self.fs.getsyspath("földér/☣.txt"))) self.fs.remove("földér/☣.txt") self.assert_not_exists("földér/☣.txt") self.fs.removedir("földér") self.assert_not_exists("földér") def test_case_sensitive(self): meta = self.fs.getmeta() if "case_insensitive" not in meta: self.skipTest("case sensitivity not known") if meta.get("case_insensitive", False): self.skipTest("the filesystem is not case sensitive.") self.fs.makedir("foo") self.fs.makedir("Foo") self.fs.touch("fOO") self.assert_exists("foo") self.assert_exists("Foo") self.assert_exists("fOO") self.assert_not_exists("FoO") self.assert_isdir("foo") self.assert_isdir("Foo") self.assert_isfile("fOO") def test_glob(self): self.assertIsInstance(self.fs.glob, glob.BoundGlobber) def test_hash(self): self.fs.makedir("foo").writebytes("hashme.txt", b"foobar" * 1024) self.assertEqual( self.fs.hash("foo/hashme.txt", "md5"), "9fff4bb103ab8ce4619064109c54cb9c" ) with self.assertRaises(errors.UnsupportedHash): self.fs.hash("foo/hashme.txt", "nohash") with self.fs.opendir("foo") as foo_fs: self.assertEqual( foo_fs.hash("hashme.txt", "md5"), "9fff4bb103ab8ce4619064109c54cb9c" )