user_repo: Add basic functionality of user's repository

Internal-tag: [#55342]
Signed-off-by: Robert Winkler <[email protected]>

pyproject.toml

@@ -22,6 +22,7 @@ dependencies = [
   "click",
   "amaranth==0.4.0.*",
   "numexpr",
+  "typing_extensions",
   "pipeline_manager_backend_communication @ git+https://github.com/antmicro/kenning-pipeline-manager-backend-communication@eb690cfb7766bfbd85a4eff2a1e809573b8b72d0",
   "pipeline_manager @ git+https://github.com/antmicro/kenning-pipeline-manager@f00e45573bb91ff5ac8760b1c4b73e8ff67085a0",
   "soc_generator @ git+https://github.com/antmicro/soc-generator@76ac5b3743bea1bb6e5d2ac7ecae31ab39d5e8b3"
@@ -44,6 +45,7 @@ tests = [
   "pytest==7.4.4",
   "pytest-cov==4.1.0",
   "pytest-lazy-fixtures==1.0.1",
+  "pyfakefs==5.3.5",
 ]
 
 [project.scripts]

tests/test_repo/__init__.py

@@ -0,0 +1,2 @@
+# Copyright (c) 2021-2024 Antmicro <www.antmicro.com>
+# SPDX-License-Identifier: Apache-2.0

tests/test_repo/test_files.py

@@ -0,0 +1,180 @@
+# Copyright (c 2024 Antmicro <www.antmicro.com>
+# SPDX-License-Identifier: Apache-2.0
+
+import filecmp
+import os
+import urllib.parse
+from pathlib import Path
+
+import pytest
+
+from topwrap.repo.files import HttpGetFile, LocalFile, TemporaryFile
+
+
+class TestFile:
+    @pytest.fixture
+    def create_file(self):
+        # A lambda is used instead of creating the file directly, because it
+        # introduces problems when the fake filesystem from pyfakefs is used
+        # in a test hierarchy. In that case, the files lifetime is not correctly
+        # managed. For hierarchical tests inherited by children, we should use
+        # lambdas to call the fakefs directly in the parent's scope.
+        return lambda: None
+
+    def skip_for_base_class(self):
+        if type(self) == TestFile:  # noqa: E721
+            pytest.skip("Skipped for base class")
+
+    @pytest.mark.usefixtures("fs")
+    def test_path_property(self, create_file):
+        self.skip_for_base_class()
+
+        file = create_file()
+        assert file.path.exists(), "The path property does not point to an existing local file"
+
+    def test_copy(self, fs, create_file):
+        self.skip_for_base_class()
+
+        file = create_file()
+        EXISTING_FILE_NAME = "existing_file.txt"
+        fs.create_file(EXISTING_FILE_NAME)
+        with pytest.raises(FileExistsError):
+            file.copy(EXISTING_FILE_NAME)
+
+        DEST_DIR_NAME = "dest_dir"
+        DEST_FILE_NAME = "dest_file.txt"
+        fs.create_dir(DEST_DIR_NAME)
+        dst = Path(DEST_DIR_NAME, DEST_FILE_NAME)
+        file.copy(dst)
+        assert dst.exists(), "The copied file hasn't been created"
+
+        assert filecmp.cmp(
+            file.path, Path(DEST_DIR_NAME, DEST_FILE_NAME), shallow=False
+        ), "The copied files don't have the same content as the original file"
+
+
+class TestTemporaryFile(TestFile):
+    @pytest.fixture()
+    def create_file(self):
+        return lambda: TemporaryFile()
+
+    @pytest.fixture()
+    def str_content(self):
+        return "My string content"
+
+    @pytest.fixture()
+    def byte_content(self):
+        return b"My Content"
+
+    @pytest.fixture()
+    def bytearray_content(self):
+        return bytearray(b"My bytearray content")
+
+    @pytest.mark.usefixtures("fs")
+    def test_create_file_without_content(self):
+        file = TemporaryFile()
+        assert file.path.exists(), "The temporary file hasn't been created"
+        read_content = file.path.read_text()
+        assert read_content == "", "The file should be empty"
+
+    @pytest.mark.usefixtures("fs")
+    def test_create_file_with_string_content(self, str_content):
+        file = TemporaryFile(str_content)
+        assert file.path.exists(), "The temporary file hasn't been created"
+        read_content = file.path.read_text()
+        assert read_content == str_content, "The file does not have expected content"
+
+    @pytest.mark.usefixtures("fs")
+    def test_create_file_with_byte_content(self, byte_content, bytearray_content):
+        file = TemporaryFile(byte_content)
+        assert file.path.exists(), "The temporary file hasn't been created"
+        read_content = file.path.read_bytes()
+        assert read_content == byte_content, "The file does not have expected content"
+
+        file = TemporaryFile(bytearray_content)
+        assert file.path.exists(), "The temporary file hasn't been created"
+        read_content = file.path.read_bytes()
+        assert (
+            bytearray(read_content) == bytearray_content
+        ), "The file does not have expected content"
+
+    @pytest.mark.usefixtures("fs")
+    def test_set_content_for_string(self, str_content):
+        file = TemporaryFile(2 * str_content)
+        file.set_content(str_content)
+        read_content = file.path.read_text()
+        assert read_content == str_content, "The file does not have expected content"
+
+    @pytest.mark.usefixtures("fs")
+    def test_set_content_for_bytes(self, byte_content, bytearray_content):
+        file = TemporaryFile(2 * byte_content)
+
+        file.set_content(byte_content)
+        read_content = file.path.read_bytes()
+        assert read_content == byte_content, "The file does not have expected content"
+
+        file.set_content(bytearray_content)
+        read_content = file.path.read_bytes()
+        assert (
+            bytearray(read_content) == bytearray_content
+        ), "The file does not have expected content"
+
+    @pytest.mark.usefixtures("fs")
+    def test_deleting_files(self):
+        file = TemporaryFile()
+        file_path = file.path
+        del file
+        os.sync()
+        assert not file_path.exists(), "The file still exists, but it should be removed"
+
+
+class TestLocalFile(TestFile):
+    @pytest.fixture()
+    def create_file(self, fs):
+        return lambda: LocalFile(fs.create_file("myfile.v").path)
+
+    @pytest.mark.usefixtures("fs")
+    def test_using_non_existing_file(self):
+        with pytest.raises(FileNotFoundError):
+            LocalFile("non_existing_file.txt")
+
+    def test_using_existing_file(self, fs):
+        file_path = Path(fs.create_file("existing_file.txt").path)
+        local_file = LocalFile(file_path)
+        assert local_file.path == file_path, "The file has an incorrect path"
+
+
+def mock_urlretrieve(url: str, file_name: str):
+    with open(file_name, "w") as fp:
+        fp.write("example content")
+
+    os.sync()
+    return (file_name, "fake headers")
+
+
+class TestHttpGetFile(TestFile):
+    @pytest.fixture()
+    def correct_url(self):
+        return "https://raw.githubusercontent.com/account/repo/master/module.v"
+
+    @pytest.fixture()
+    def create_file(self, fs, monkeypatch, correct_url):
+        monkeypatch.setattr(urllib.request, "urlretrieve", mock_urlretrieve)
+        return lambda: HttpGetFile(correct_url)
+
+    @pytest.mark.usefixtures("fs")
+    def test_wrong_url(self, monkeypatch):
+        monkeypatch.setattr(urllib.request, "urlretrieve", mock_urlretrieve)
+        with pytest.raises(Exception, match=r"Url .* doesn't seem to have a correct path.*"):
+            HttpGetFile("wrong url")
+
+    def test_download_dir(self, fs, monkeypatch, correct_url):
+        DOWNLOAD_DIR = "download_dir"
+        fs.create_dir(DOWNLOAD_DIR)
+        download_dir = Path(DOWNLOAD_DIR)
+
+        http_file = HttpGetFile(correct_url, download_dir)
+        monkeypatch.setattr(urllib.request, "urlretrieve", mock_urlretrieve)
+        http_file.download()
+        output_path = download_dir / "module.v"
+        assert output_path.exists(), "The file hasn't been created"

tests/test_repo/test_repo.py

@@ -0,0 +1,78 @@
+# Copyright (c) 2024 Antmicro <www.antmicro.com>
+# SPDX-License-Identifier: Apache-2.0
+
+
+import pytest
+
+from topwrap.repo.files import TemporaryFile
+from topwrap.repo.repo import Repo, ResourceNotSupportedException
+from topwrap.repo.resource import FileHandler, Resource, ResourceHandler
+
+
+class MySupportedResource(Resource):
+    pass
+
+
+class MyUnsupportedResource(Resource):
+    pass
+
+
+class MyFileHandler(FileHandler):
+    resource_type = Resource
+
+    def __init__(self, files):
+        self._files = files
+
+    @property
+    def files(self):
+        return self._files
+
+    def parse(self):
+        return [MySupportedResource()]
+
+
+class MyWrongFileHandler(FileHandler):
+    resource_type = Resource
+
+    def __init__(self, files):
+        self._files = files
+
+    @property
+    def files(self):
+        return self._files
+
+    def parse(self):
+        return [MyUnsupportedResource()]
+
+
+class MySupportedResourceHandler(ResourceHandler):
+    resource_type = MySupportedResource
+
+    def save(self, resource, repo_path):
+        pass
+
+    def load(self, repo_path):
+        return []
+
+
+class MyRepo(Repo):
+    def __init__(self2):
+        resource_handlers = [MySupportedResourceHandler()]
+        super().__init__(resource_handlers)
+
+
+class TestRepoBase:
+    @pytest.fixture()
+    def repo(self):
+        return MyRepo()
+
+    @pytest.mark.usefixtures("fs")
+    def test_save_supported_resource(self, repo):
+        repo.add_files(MyFileHandler([TemporaryFile()]))
+        repo.save("repo")
+
+    @pytest.mark.usefixtures("fs")
+    def test_save_unsupported_resource(self, repo):
+        repo.add_files(MyWrongFileHandler([TemporaryFile()]))
+        with pytest.raises(ResourceNotSupportedException):
+            repo.save("repo")