fix(registry): module typed (#811)

Author: Tranquility2

modules/registry/testcontainers/registry/__init__.py

@@ -1,7 +1,7 @@
 import time
 from io import BytesIO
 from tarfile import TarFile, TarInfo
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Any, Optional
 
 import bcrypt
 from requests import get
@@ -25,7 +25,7 @@ def __init__(
         port: int = 5000,
         username: Optional[str] = None,
         password: Optional[str] = None,
-        **kwargs,
+        **kwargs: Any,
     ) -> None:
         super().__init__(image=image, **kwargs)
         self.port: int = port
@@ -35,6 +35,8 @@ def __init__(
 
     def _copy_credentials(self) -> None:
         # Create credentials and write them to the container
+        if self.password is None:
+            raise ValueError("Password cannot be None")
         hashed_password: str = bcrypt.hashpw(
             self.password.encode("utf-8"),
             bcrypt.gensalt(rounds=12, prefix=b"2a"),
@@ -44,7 +46,7 @@ def _copy_credentials(self) -> None:
         with BytesIO() as tar_archive_object, TarFile(fileobj=tar_archive_object, mode="w") as tmp_tarfile:
             tarinfo: TarInfo = TarInfo(name=self.credentials_path)
             tarinfo.size = len(content)
-            tarinfo.mtime = time.time()
+            tarinfo.mtime = int(time.time())
 
             tmp_tarfile.addfile(tarinfo, BytesIO(content))
             tar_archive_object.seek(0)
@@ -54,12 +56,13 @@ def _copy_credentials(self) -> None:
     def _readiness_probe(self) -> None:
         url: str = f"http://{self.get_registry()}/v2"
         if self.username and self.password:
-            response: Response = get(url, auth=HTTPBasicAuth(self.username, self.password), timeout=1)
+            auth_response: Response = get(url, auth=HTTPBasicAuth(self.username, self.password), timeout=1)
+            auth_response.raise_for_status()
         else:
             response: Response = get(url, timeout=1)
-        response.raise_for_status()
+            response.raise_for_status()
 
-    def start(self):
+    def start(self) -> "DockerRegistryContainer":
         if self.username and self.password:
             self.with_env("REGISTRY_AUTH_HTPASSWD_REALM", "local-registry")
             self.with_env("REGISTRY_AUTH_HTPASSWD_PATH", self.credentials_path)

modules/registry/tests/test_registry.py

@@ -7,7 +7,7 @@
 REGISTRY_PASSWORD: str = "bar"
 
 
-def test_registry():
+def test_registry() -> None:
     with DockerRegistryContainer().with_bind_ports(5000, 5000) as registry_container:
         url: str = f"http://{registry_container.get_registry()}/v2/_catalog"
 
@@ -16,7 +16,7 @@ def test_registry():
         assert response.status_code == 200
 
 
-def test_registry_with_authentication():
+def test_registry_with_authentication() -> None:
     with DockerRegistryContainer(username=REGISTRY_USERNAME, password=REGISTRY_PASSWORD).with_bind_ports(
         5000, 5000
     ) as registry_container:

pyproject.toml

@@ -345,6 +345,11 @@ module = ['wrapt.*']
 # wrapt doesn't have type annotations
 ignore_missing_imports = true
 
+[[tool.mypy.overrides]]
+module = ['requests.*']
+# requests doesn't have type annotations
+ignore_missing_imports = true
+
 [build-system]
 requires = ["poetry-core"]
 build-backend = "poetry.core.masonry.api"