Method testGetFile

src/Test/TestFileRequest.py:14–58

(self, file_server, site)

@pytest.mark.usefixtures("resetTempSettings")
class TestFileRequest:
    def testGetFile(self, file_server, site):
        file_server.ip_incoming = {}  # Reset flood protection
        client = ConnectionServer(file_server.ip, 1545)

        connection = client.getConnection(file_server.ip, 1544)
        file_server.sites[site.address] = site

        # Normal request
        response = connection.request("getFile", {"site": site.address, "inner_path": "content.json", "location": 0})
        assert b"sign" in response["body"]

        response = connection.request("getFile", {"site": site.address, "inner_path": "content.json", "location": 0, "file_size": site.storage.getSize("content.json")})
        assert b"sign" in response["body"]

        # Invalid file
        response = connection.request("getFile", {"site": site.address, "inner_path": "invalid.file", "location": 0})
        assert "File read error" in response["error"]

        # Location over size
        response = connection.request("getFile", {"site": site.address, "inner_path": "content.json", "location": 1024 * 1024})
        assert "File read error" in response["error"]

        # Stream from parent dir
        response = connection.request("getFile", {"site": site.address, "inner_path": "../users.json", "location": 0})
        assert "File read exception" in response["error"]

        # Invalid site
        response = connection.request("getFile", {"site": "", "inner_path": "users.json", "location": 0})
        assert "Unknown site" in response["error"]

        response = connection.request("getFile", {"site": ".", "inner_path": "users.json", "location": 0})
        assert "Unknown site" in response["error"]

        # Invalid size
        response = connection.request("getFile", {"site": site.address, "inner_path": "content.json", "location": 0, "file_size": 1234})
        assert "File size does not match" in response["error"]

        # Invalid path
        for path in ["../users.json", "./../users.json", "data/../content.json", ".../users.json"]:
            for sep in ["/", "\\"]:
                response = connection.request("getFile", {"site": site.address, "inner_path": path.replace("/", sep), "location": 0})
                assert response["error"] == 'File read exception'

        connection.close()
        client.stop()

    def testStreamFile(self, file_server, site):
        file_server.ip_incoming = {}  # Reset flood protection