cleanup and add test_backoffice

backoffice/__init__.py

@@ -0,0 +1 @@
+from ._backoffice import BackOffice

backoffice/__main__.py

@@ -1,54 +1,6 @@
-import os
-from typing import Literal, Optional
-
 import fire
-from bioimageio.spec.model.v0_5 import WeightsFormat
-from dotenv import load_dotenv
-
-from .backup import backup
-from .run_dynamic_tests import run_dynamic_tests
-from .utils.remote_resource import PublishedVersion, RemoteResource, StagedVersion
-from .utils.s3_client import Client
-from .validate_format import validate_format
-
-_ = load_dotenv()
-
-
-class BackOffice:
-    def __init__(self) -> None:
-        super().__init__()
-        self.client = Client()
-
-    def stage(self, resource_id: str, package_url: str):
-        resource = RemoteResource(client=Client(), id=resource_id)
-        staged = resource.stage_new_version(package_url)
-        validate_format(staged)
-
-    def test(
-        self,
-        resource_id: str,
-        version: int,
-        weight_format: Optional[WeightsFormat] = None,
-        create_env_outcome: Literal["success", ""] = "success",
-    ):
-        staged = StagedVersion(self.client, resource_id, version)
-        run_dynamic_tests(
-            staged=staged,
-            weight_format=weight_format,
-            create_env_outcome=create_env_outcome,
-        )
-
-    def await_review(self, resource_id: str, version: int):
-        staged = StagedVersion(self.client, resource_id, version)
-        staged.await_review()
-
-    def publish(self, resource_id: str, stage_nr: int):
-        staged = StagedVersion(client=self.client, id=resource_id, version=stage_nr)
-        published = staged.publish()
-        assert isinstance(published, PublishedVersion)
 
-    def backup(self):
-        _ = backup(self.client, os.environ["ZENODO_URL"])
+from backoffice import BackOffice
 
 
 def main():

backoffice/_backoffice.py

@@ -0,0 +1,59 @@
+import os
+from typing import Literal, Optional
+
+from bioimageio.spec.model.v0_5 import WeightsFormat
+from dotenv import load_dotenv
+
+from backoffice.backup import backup
+from backoffice.run_dynamic_tests import run_dynamic_tests
+from backoffice.utils.remote_resource import (
+    PublishedVersion,
+    RemoteResource,
+    StagedVersion,
+)
+from backoffice.utils.s3_client import Client
+from backoffice.validate_format import validate_format
+
+_ = load_dotenv()
+
+
+class BackOffice:
+    def __init__(
+        self,
+        host: str = os.environ["S3_HOST"],
+        bucket: str = os.environ["S3_BUCKET"],
+        prefix: str = os.environ["S3_FOLDER"],
+    ) -> None:
+        super().__init__()
+        self.client = Client(host=host, bucket=bucket, prefix=prefix)
+
+    def stage(self, resource_id: str, package_url: str):
+        resource = RemoteResource(client=Client(), id=resource_id)
+        staged = resource.stage_new_version(package_url)
+        validate_format(staged)
+
+    def test(
+        self,
+        resource_id: str,
+        version: int,
+        weight_format: Optional[WeightsFormat] = None,
+        create_env_outcome: Literal["success", ""] = "success",
+    ):
+        staged = StagedVersion(self.client, resource_id, version)
+        run_dynamic_tests(
+            staged=staged,
+            weight_format=weight_format,
+            create_env_outcome=create_env_outcome,
+        )
+
+    def await_review(self, resource_id: str, stage_nr: int):
+        staged = StagedVersion(self.client, resource_id, stage_nr)
+        staged.await_review()
+
+    def publish(self, resource_id: str, stage_nr: int):
+        staged = StagedVersion(self.client, resource_id, stage_nr)
+        published = staged.publish()
+        assert isinstance(published, PublishedVersion)
+
+    def backup(self, destination: str = os.environ["ZENODO_URL"]):
+        _ = backup(self.client, destination)

backoffice/backup.py

@@ -1,5 +1,3 @@
-import os
-
 from dotenv import load_dotenv
 from loguru import logger
 

tests/conftest.py

@@ -1,23 +1,33 @@
 import os
-from typing import TYPE_CHECKING
 
 import pytest
 
-if TYPE_CHECKING:
-    from scripts.utils.s3_client import Client
+from backoffice import BackOffice
+from backoffice.utils.s3_client import Client
 
 
 @pytest.fixture(scope="session")
-def client():
-    from scripts.utils.s3_client import Client
+def backoffice():
+    bo = BackOffice(
+        host=os.environ["S3_HOST"],
+        bucket=os.environ["S3_TEST_BUCKET"],
+        prefix=os.environ["S3_TEST_FOLDER"] + "/pytest/backoffice",
+    )
+    bo.client.rm_dir("")  # wipe s3 test folder
+    yield bo
+    bo.client.rm_dir("")  # wipe s3 test folder
 
+
+@pytest.fixture(scope="session")
+def client():
     cl = Client(
         host=os.environ["S3_HOST"],
         bucket=os.environ["S3_TEST_BUCKET"],
-        prefix=os.environ["S3_TEST_FOLDER"] + "/pytest",
+        prefix=os.environ["S3_TEST_FOLDER"] + "/pytest/client",
     )
     cl.rm_dir("")  # wipe s3 test folder
     yield cl
+    cl.rm_dir("")  # wipe s3 test folder
 
 
 @pytest.fixture(scope="session")

tests/test_backoffice.py

@@ -0,0 +1,12 @@
+import os
+
+from backoffice import BackOffice
+
+
+def test_backoffice(
+    backoffice: BackOffice, package_url: str, package_id: str, s3_test_folder_url: str
+):
+    backoffice.stage(resource_id=package_id, package_url=package_url)
+    backoffice.await_review(resource_id=package_id, stage_nr=1)
+    backoffice.publish(resource_id=package_id, stage_nr=1)
+    backoffice.backup(os.environ["ZENODO_TEST_URL"])