Merge pull request #54 from canonical/add-type-hints

Add type hints

lib/charms/kratos/v0/kratos_endpoints.py

@@ -52,7 +52,7 @@ def some_event_function():
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 2
+LIBPATCH = 3
 
 RELATION_NAME = "kratos-endpoint-info"
 INTERFACE_NAME = "kratos_endpoints"
@@ -139,7 +139,10 @@ def get_kratos_endpoints(self) -> Dict:
         if len(endpoints) == 0:
             raise KratosEndpointsRelationMissingError()
 
-        data = endpoints[0].data[endpoints[0].app]
+        if not (app := endpoints[0].app):
+            raise KratosEndpointsRelationMissingError()
+
+        data = endpoints[0].data[app]
 
         if "public_endpoint" not in data:
             raise KratosEndpointsRelationDataMissingError(

pyproject.toml

@@ -38,3 +38,25 @@ docstring-convention = "google"
 copyright-check = "True"
 copyright-author = "Canonical Ltd."
 copyright-regexp = "Copyright\\s\\d{4}([-,]\\d{4})*\\s+%(author)s"
+
+[tool.mypy]
+pretty = true
+mypy_path = "./src:./lib/:./tests"
+# Exclude non-kratos libraries
+exclude = 'lib/charms/((?!kratos/).)'
+follow_imports = "silent"
+warn_redundant_casts = true
+warn_unused_configs = true
+show_traceback = true
+show_error_codes = true
+namespace_packages = true
+explicit_package_bases = true
+check_untyped_defs = true
+allow_redefinition = true
+disallow_incomplete_defs = true
+disallow_untyped_defs = true
+
+# Ignore libraries that do not have type hint nor stubs
+[[tool.mypy.overrides]]
+module = ["ops.*", "pytest.*", "pytest_operator.*", "urllib3.*", "jinja2.*", "lightkube.*", "pytest_mock.*"]
+ignore_missing_imports = true

src/charm.py

@@ -10,10 +10,11 @@
 from functools import cached_property
 from os.path import join
 from pathlib import Path
-from typing import Dict, Optional
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
 
 import requests
 from charms.data_platform_libs.v0.data_interfaces import (
+    DatabaseCreatedEvent,
     DatabaseEndpointsChangedEvent,
     DatabaseRequires,
 )
@@ -33,13 +34,24 @@
     IngressPerAppRevokedEvent,
 )
 from jinja2 import Template
-from ops.charm import ActionEvent, CharmBase, HookEvent, RelationEvent
+from ops.charm import (
+    ActionEvent,
+    CharmBase,
+    ConfigChangedEvent,
+    HookEvent,
+    PebbleReadyEvent,
+    RelationEvent,
+)
 from ops.main import main
 from ops.model import ActiveStatus, BlockedStatus, MaintenanceStatus, ModelError, WaitingStatus
 from ops.pebble import Error, ExecError, Layer
 
 from kratos import KratosAPI
 
+if TYPE_CHECKING:
+    from ops.pebble import LayerDict
+
+
 logger = logging.getLogger(__name__)
 KRATOS_ADMIN_PORT = 4434
 KRATOS_PUBLIC_PORT = 4433
@@ -62,7 +74,7 @@ def dict_to_action_output(d: Dict) -> Dict:
 class KratosCharm(CharmBase):
     """Charmed Ory Kratos."""
 
-    def __init__(self, *args):
+    def __init__(self, *args: Any) -> None:
         super().__init__(*args)
         self._container_name = "kratos"
         self._container = self.unit.get_container(self._container_name)
@@ -140,7 +152,7 @@ def __init__(self, *args):
         self.framework.observe(self.on.run_migration_action, self._on_run_migration_action)
 
     @property
-    def _kratos_service_params(self):
+    def _kratos_service_params(self) -> str:
         ret = ["--config", str(self._config_file_path)]
         if self.config["dev"]:
             logger.warning("Running Kratos in dev mode, don't do this in production")
@@ -161,7 +173,7 @@ def _kratos_service_is_running(self) -> bool:
 
     @property
     def _pebble_layer(self) -> Layer:
-        pebble_layer = {
+        pebble_layer: LayerDict = {
             "summary": "kratos layer",
             "description": "pebble config layer for kratos",
             "services": {
@@ -186,11 +198,11 @@ def _pebble_layer(self) -> Layer:
         return Layer(pebble_layer)
 
     @property
-    def _domain_url(self):
+    def _domain_url(self) -> Optional[str]:
         return self.config["external_url"] or self.public_ingress.url
 
     @cached_property
-    def _get_available_mappers(self):
+    def _get_available_mappers(self) -> List[str]:
         return [
             schema_file.name[: -len("_schema.jsonnet")]
             for schema_file in self._mappers_local_dir_path.iterdir()
@@ -219,7 +231,7 @@ def _render_conf_file(self) -> str:
         )
         return rendered
 
-    def _push_schemas(self):
+    def _push_schemas(self) -> None:
         for schema_file in self._mappers_local_dir_path.iterdir():
             with open(Path(schema_file)) as f:
                 schema = f.read()
@@ -271,7 +283,7 @@ def _run_sql_migration(self) -> bool:
             stdout, _ = process.wait_output()
             logger.info(f"Successfully executed automigration: {stdout}")
         except ExecError as err:
-            logger.error(f"Exited with code {err.exit_code}. Stderr: {err.stderr}")
+            logger.error(f"Exited with code {err.exit_code}. Stderr: {err.stderr!r}")
             self.unit.status = BlockedStatus("Database migration job failed")
             return False
 
@@ -305,7 +317,7 @@ def _handle_status_update_config(self, event: HookEvent) -> None:
         else:
             self.unit.status = BlockedStatus("Missing postgres database relation")
 
-    def _on_pebble_ready(self, event) -> None:
+    def _on_pebble_ready(self, event: PebbleReadyEvent) -> None:
         """Event Handler for pebble ready event."""
         if not self._container.can_connect():
             event.defer()
@@ -346,7 +358,7 @@ def _on_pebble_ready(self, event) -> None:
         else:
             self.unit.status = BlockedStatus("Missing postgres database relation")
 
-    def _on_config_changed(self, event) -> None:
+    def _on_config_changed(self, event: ConfigChangedEvent) -> None:
         """Event Handler for config changed event."""
         self._handle_status_update_config(event)
 
@@ -368,7 +380,7 @@ def _update_kratos_endpoints_relation_data(self, event: RelationEvent) -> None:
 
         self.endpoints_provider.send_endpoint_relation_data(admin_endpoint[0], public_endpoint[0])
 
-    def _on_database_created(self, event) -> None:
+    def _on_database_created(self, event: DatabaseCreatedEvent) -> None:
         """Event Handler for database created event."""
         if not self._container.can_connect():
             event.defer()

tests/integration/test_charm.py

@@ -28,7 +28,7 @@ async def get_unit_address(ops_test: OpsTest, app_name: str, unit_num: int) -> s
 
 
 @pytest.mark.abort_on_fail
-async def test_build_and_deploy(ops_test: OpsTest):
+async def test_build_and_deploy(ops_test: OpsTest) -> None:
     """Build the charm-under-test and deploy it.
 
     Assert on the unit status before any relations/configurations take place.
@@ -55,7 +55,7 @@ async def test_build_and_deploy(ops_test: OpsTest):
         assert ops_test.model.applications[APP_NAME].units[0].workload_status == "active"
 
 
-async def test_ingress_relation(ops_test: OpsTest):
+async def test_ingress_relation(ops_test: OpsTest) -> None:
     await ops_test.model.deploy(
         TRAEFIK,
         application_name=TRAEFIK_PUBLIC_APP,
@@ -79,7 +79,7 @@ async def test_ingress_relation(ops_test: OpsTest):
     )
 
 
-async def test_has_public_ingress(ops_test: OpsTest):
+async def test_has_public_ingress(ops_test: OpsTest) -> None:
     # Get the traefik address and try to reach kratos
     public_address = await get_unit_address(ops_test, TRAEFIK_PUBLIC_APP, 0)
 
@@ -90,7 +90,7 @@ async def test_has_public_ingress(ops_test: OpsTest):
     assert resp.status_code == 200
 
 
-async def test_has_admin_ingress(ops_test: OpsTest):
+async def test_has_admin_ingress(ops_test: OpsTest) -> None:
     # Get the traefik address and try to reach kratos
     admin_address = await get_unit_address(ops_test, TRAEFIK_ADMIN_APP, 0)
 

tests/unit/conftest.py

@@ -6,6 +6,7 @@
 from unittest.mock import MagicMock
 
 import pytest
+from ops.model import Container
 from ops.pebble import ExecError
 from ops.testing import Harness
 from pytest_mock import MockerFixture
@@ -15,7 +16,7 @@
 
 
 @pytest.fixture()
-def harness(mocked_kubernetes_service_patcher) -> Harness:
+def harness(mocked_kubernetes_service_patcher: MagicMock) -> Harness:
     harness = Harness(KratosCharm)
     harness.set_model_name("kratos-model")
     harness.set_can_connect("kratos", True)
@@ -32,17 +33,17 @@ def mocked_kratos_process() -> MagicMock:
 
 
 @pytest.fixture()
-def kratos_api(mocked_kratos_process):
+def kratos_api(mocked_kratos_process: MagicMock) -> KratosAPI:
     container = MagicMock()
     container.exec = MagicMock(return_value=mocked_kratos_process)
     return KratosAPI("http://localhost:4434", container, "/etc/config/kratos.yaml")
 
 
 @pytest.fixture()
-def mocked_kubernetes_service_patcher(mocker):
+def mocked_kubernetes_service_patcher(mocker: MockerFixture) -> MagicMock:
     mocked_service_patcher = mocker.patch("charm.KubernetesServicePatch")
     mocked_service_patcher.return_value = lambda x, y: None
-    yield mocked_service_patcher
+    return mocked_service_patcher
 
 
 @pytest.fixture()
@@ -54,39 +55,39 @@ def mocked_kratos_service(harness: Harness, mocked_container: MagicMock) -> Gene
 
 
 @pytest.fixture()
-def mocked_fqdn(mocker):
+def mocked_fqdn(mocker: MockerFixture) -> MagicMock:
     mocked_fqdn = mocker.patch("socket.getfqdn")
     mocked_fqdn.return_value = "kratos"
     return mocked_fqdn
 
 
 @pytest.fixture()
-def mocked_container(harness, mocker):
+def mocked_container(harness: Harness, mocker: MockerFixture) -> Container:
     container = harness.model.unit.get_container("kratos")
-    container.restart = mocker.MagicMock()
+    setattr(container, "restart", mocker.MagicMock())
     return container
 
 
 @pytest.fixture()
-def mocked_pebble_exec(mocker):
+def mocked_pebble_exec(mocker: MockerFixture) -> MagicMock:
     mocked_pebble_exec = mocker.patch("ops.model.Container.exec")
-    yield mocked_pebble_exec
+    return mocked_pebble_exec
 
 
 @pytest.fixture()
-def mocked_pebble_exec_success(mocker, mocked_pebble_exec):
+def mocked_pebble_exec_success(mocker: MockerFixture, mocked_pebble_exec: MagicMock) -> MagicMock:
     mocked_process = mocker.patch("ops.pebble.ExecProcess")
     mocked_process.wait_output.return_value = ("Success", None)
     mocked_pebble_exec.return_value = mocked_process
-    yield mocked_pebble_exec
+    return mocked_pebble_exec
 
 
 @pytest.fixture()
-def mocked_pebble_exec_failed(mocked_pebble_exec):
+def mocked_pebble_exec_failed(mocked_pebble_exec: MagicMock) -> MagicMock:
     mocked_pebble_exec.side_effect = ExecError(
-        exit_code=400, stderr="Failed to execute", stdout="Failed", command="test command"
+        exit_code=400, stderr="Failed to execute", stdout="Failed", command=["test", "command"]
     )
-    yield
+    return mocked_pebble_exec
 
 
 @pytest.fixture()