diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml index 448f159..1d67a58 100644 --- a/.github/workflows/build.yaml +++ b/.github/workflows/build.yaml @@ -10,8 +10,6 @@ jobs: arch: - arch: amd64 runner: ubuntu-22.04 - - arch: arm64 - runner: [self-hosted, linux, ARM64, medium, jammy] runs-on: ${{ matrix.arch.runner }} steps: - uses: actions/checkout@v4 diff --git a/.github/workflows/publish-charm.yaml b/.github/workflows/publish-charm.yaml index 7fd25db..0e24778 100644 --- a/.github/workflows/publish-charm.yaml +++ b/.github/workflows/publish-charm.yaml @@ -13,8 +13,7 @@ jobs: arch: - arch: amd64 runner: ubuntu-22.04 - - arch: arm64 - runner: [self-hosted, linux, ARM64, medium, jammy] + runs-on: ${{ matrix.arch.runner }} steps: - name: Checkout @@ -33,7 +32,7 @@ jobs: run: echo "charm_path=$(find . -name '*.charm' -type f -print)" >> $GITHUB_OUTPUT - name: Upload charm to Charmhub - uses: canonical/charming-actions/upload-charm@2.6.2 + uses: canonical/charming-actions/upload-charm@2.6.3 with: built-charm-path: ${{ steps.charm-path.outputs.charm_path }} credentials: "${{ secrets.CHARMCRAFT_AUTH }}" diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 1f8c192..129bde3 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -34,5 +34,5 @@ charmcraft pack Deploy it by using: ```shell -juju deploy ./gocert-k8s_ubuntu-22.04-amd64.charm --resource gocert-image=ghcr.io/canonical/gocert +juju deploy ./notary-k8s_ubuntu-22.04-amd64.charm --resource notary-image=ghcr.io/canonical/notary ``` diff --git a/README.md b/README.md index f70bd60..ae80158 100644 --- a/README.md +++ b/README.md @@ -1,20 +1,20 @@ -# gocert-k8s-operator +# notary-k8s-operator Manage your certificates in charms ## OCI Images -- GoCert: [ghcr.io/canonical/gocert](https://github.com/canonical/gocert) +- Notary: [ghcr.io/canonical/notary](https://github.com/canonical/notary) -# gocert +# notary -Charmhub package name: gocert -More information: https://charmhub.io/gocert +Charmhub package name: notary +More information: https://charmhub.io/notary -GoCert helps you manage your certificates, from simple all the way up to enterprise deployments. +Notary helps you manage your certificates, from simple all the way up to enterprise deployments. ## Other resources -- [Read more](https://github.com/canonical/gocert/blob/main/README.md) +- [Read more](https://github.com/canonical/notary/blob/main/README.md) - [Contributing](CONTRIBUTING.md) diff --git a/charmcraft.yaml b/charmcraft.yaml index 2d73458..3e5334a 100644 --- a/charmcraft.yaml +++ b/charmcraft.yaml @@ -1,23 +1,26 @@ -name: gocert-k8s +name: notary-k8s type: charm -title: GoCert +title: Notary summary: Certificate management made easy description: | - GoCert helps you manage certificate requests and their associated certificates. - Charmed GoCert helps you automatically receive CSRs and distribute certificates to the applications + Notary helps you manage certificate requests and their associated certificates. + Charmed Notary helps you automatically receive CSRs and distribute certificates to the applications you've deployed in your model. links: documentation: https://discourse.charmhub.io/t/gocert-docs-index/15216 issues: - - https://github.com/canonical/gocert-k8s-operator/issues + - https://github.com/canonical/notary-k8s-operator/issues source: - - https://github.com/canonical/gocert-k8s-operator + - https://github.com/canonical/notary-k8s-operator website: - - https://charmhub.io/gocert-k8s + - https://charmhub.io/notary-k8s +base: ubuntu@24.04 +platforms: + amd64: provides: certificates: @@ -34,22 +37,14 @@ requires: interface: ingress limit: 1 -bases: - - build-on: - - name: ubuntu - channel: "22.04" - run-on: - - name: ubuntu - channel: "22.04" - containers: - gocert: - resource: gocert-image + notary: + resource: notary-image mounts: - storage: config - location: /etc/gocert/config + location: /etc/notary/config - storage: database - location: /var/lib/gocert/database + location: /var/lib/notary/database storage: config: @@ -60,10 +55,10 @@ storage: minimum-size: 1G resources: - gocert-image: + notary-image: type: oci-image - description: OCI image for the 'GoCert' application - upstream-source: ghcr.io/canonical/gocert:0.0.3 + description: OCI image for the Notary application + upstream-source: ghcr.io/canonical/notary:0.0.3 parts: charm: diff --git a/requirements.txt b/requirements.txt index 811910c..b64ba61 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,38 +6,61 @@ # annotated-types==0.7.0 # via pydantic +anyio==4.4.0 + # via httpx certifi==2024.7.4 - # via requests + # via + # httpcore + # httpx + # requests cffi==1.17.0 # via cryptography charset-normalizer==3.3.2 # via requests -cosl==0.0.23 +cosl==0.0.33 # via -r requirements.in -cryptography==43.0.0 +cryptography==43.0.1 # via -r requirements.in +h11==0.14.0 + # via httpcore +httpcore==1.0.5 + # via httpx +httpx==0.27.2 + # via lightkube idna==3.7 - # via requests -ops==2.15.0 + # via + # anyio + # httpx + # requests +lightkube==0.15.4 + # via cosl +lightkube-models==1.30.0.8 + # via lightkube +ops==2.16.1 # via # -r requirements.in # cosl pycparser==2.22 # via cffi -pydantic==2.8.2 +pydantic==2.9.1 # via # -r requirements.in # cosl -pydantic-core==2.20.1 +pydantic-core==2.23.3 # via pydantic pyyaml==6.0.2 # via # cosl + # lightkube # ops requests==2.32.3 # via -r requirements.in rpds-py==0.18.0 # via -r requirements.in +sniffio==1.3.1 + # via + # anyio + # httpx tenacity==9.0.0 # via cosl typing-extensions==4.12.2 diff --git a/src/charm.py b/src/charm.py index b709e6a..6ba8891 100755 --- a/src/charm.py +++ b/src/charm.py @@ -11,11 +11,13 @@ from dataclasses import dataclass import ops +import yaml from charms.grafana_k8s.v0.grafana_dashboard import GrafanaDashboardProvider from charms.loki_k8s.v1.loki_push_api import LogForwarder from charms.prometheus_k8s.v0.prometheus_scrape import MetricsEndpointProvider from charms.tls_certificates_interface.v4.tls_certificates import ( Certificate, + ProviderCertificate, TLSCertificatesProvidesV4, generate_ca, generate_certificate, @@ -24,10 +26,12 @@ ) from charms.traefik_k8s.v2.ingress import IngressPerAppRequirer -from gocert import GoCert +from notary import Notary logger = logging.getLogger(__name__) +CERTIFICATE_PROVIDER_RELATION_NAME = "certificates" + LOGGING_RELATION_NAME = "logging" METRICS_RELATION_NAME = "metrics" GRAFANA_RELATION_NAME = "grafana-dashboard" @@ -35,16 +39,17 @@ DB_MOUNT = "database" CONFIG_MOUNT = "config" CHARM_PATH = "/var/lib/juju/storage" -WORKLOAD_CONFIG_PATH = "/etc/gocert" +WORKLOAD_CONFIG_PATH = "/etc/notary" +WORKLOAD_DB_PATH = "/var/lib" -CERTIFICATE_COMMON_NAME = "GoCert Self Signed Certificate" -SELF_SIGNED_CA_COMMON_NAME = "GoCert Self Signed Root CA" -GOCERT_LOGIN_SECRET_LABEL = "GoCert Login Details" +CERTIFICATE_COMMON_NAME = "Notary Self Signed Certificate" +SELF_SIGNED_CA_COMMON_NAME = "Notary Self Signed Root CA" +NOTARY_LOGIN_SECRET_LABEL = "Notary Login Details" @dataclass class LoginSecret: - """The format of the secret for the login details that are required to login to GoCert.""" + """The format of the secret for the login details that are required to login to Notary.""" username: str password: str @@ -59,16 +64,18 @@ def to_dict(self) -> dict[str, str]: } -class GocertCharm(ops.CharmBase): - """Charmed Gocert.""" +class NotaryCharm(ops.CharmBase): + """Charmed Notary.""" def __init__(self, framework: ops.Framework): super().__init__(framework) self.port = 2111 self.unit.set_ports(self.port) - self.container = self.unit.get_container("gocert") - self.tls = TLSCertificatesProvidesV4(self, relationship_name="certificates") + self.container = self.unit.get_container("notary") + self.tls = TLSCertificatesProvidesV4( + self, relationship_name=CERTIFICATE_PROVIDER_RELATION_NAME + ) self.dashboard = GrafanaDashboardProvider(self, relation_name=GRAFANA_RELATION_NAME) self.logs = LogForwarder(charm=self, relation_name=LOGGING_RELATION_NAME) self.ingress = IngressPerAppRequirer( @@ -90,15 +97,15 @@ def __init__(self, framework: ops.Framework): ], ) - self.client = GoCert( + self.client = Notary( f"https://{self._application_bind_address}:{self.port}", f"{CHARM_PATH}/{CONFIG_MOUNT}/0/ca.pem", ) [ framework.observe(event, self.configure) for event in [ - self.on["gocert"].pebble_ready, - self.on["gocert"].pebble_custom_notice, + self.on["notary"].pebble_ready, + self.on["notary"].pebble_custom_notice, self.on["certificates"].relation_changed, self.on.config_storage_attached, self.on.database_storage_attached, @@ -116,9 +123,10 @@ def configure(self, event: ops.EventBase): return if not self.container.can_connect(): return - self._configure_gocert_config_file() + self._configure_notary_config_file() self._configure_access_certificates() self._configure_charm_authorization() + self._configure_certificate_requirers() def _on_collect_status(self, event: ops.CollectStatusEvent): if not self.unit.is_leader(): @@ -134,61 +142,136 @@ def _on_collect_status(self, event: ops.CollectStatusEvent): event.add_status(ops.WaitingStatus("certificates not yet created")) return if not self.client.is_api_available(): - event.add_status(ops.WaitingStatus("GoCert server not yet available")) + event.add_status(ops.WaitingStatus("Notary server not yet available")) return if not self.client.is_initialized(): - event.add_status(ops.BlockedStatus("Please initialize GoCert")) + event.add_status(ops.BlockedStatus("Please initialize Notary")) return event.add_status(ops.ActiveStatus()) ## Configure Dependencies ## - def _configure_gocert_config_file(self): + def _configure_notary_config_file(self): """Push the config file.""" try: self.container.pull(f"{WORKLOAD_CONFIG_PATH}/config/config.yaml") logger.info("Config file already created.") except ops.pebble.PathError: - config_file = open("src/config/config.yaml").read() self.container.make_dir(path=f"{WORKLOAD_CONFIG_PATH}/config", make_parents=True) self.container.push( - path=f"{WORKLOAD_CONFIG_PATH}/config/config.yaml", source=config_file + path=f"{WORKLOAD_CONFIG_PATH}/config/config.yaml", + source=yaml.dump( + data={ + "key_path": f"{WORKLOAD_CONFIG_PATH}/config/private_key.pem", + "cert_path": f"{WORKLOAD_CONFIG_PATH}/config/certificate.pem", + "db_path": f"{WORKLOAD_DB_PATH}/notary/database/certs.db", + "port": self.port, + "pebble_notifications": True, + } + ), ) logger.info("Config file created.") def _configure_access_certificates(self): - """Update the config files for gocert and replan if required.""" + """Update the config files for notary and replan if required.""" certificates_changed = False if not self._self_signed_certificates_generated(): certificates_changed = True self._generate_self_signed_certificates() logger.info("Certificates configured.") if certificates_changed: - self.container.add_layer("gocert", self._pebble_layer, combine=True) + self.container.add_layer("notary", self._pebble_layer, combine=True) with suppress(ops.pebble.ChangeError): self.container.replan() def _configure_charm_authorization(self): - """Create an admin user to manage GoCert if needed, and acquire a token by logging in if needed.""" + """Create an admin user to manage Notary if needed, and acquire a token by logging in if needed.""" login_details = self._get_or_create_admin_account() if not login_details: return if not login_details.token or not self.client.token_is_valid(login_details.token): login_details.token = self.client.login(login_details.username, login_details.password) - login_details_secret = self.model.get_secret(label=GOCERT_LOGIN_SECRET_LABEL) + login_details_secret = self.model.get_secret(label=NOTARY_LOGIN_SECRET_LABEL) login_details_secret.set_content(login_details.to_dict()) + def _configure_certificate_requirers(self): + """Get all CSR's and certs from databags and Notary, compare differences and update requirers if needed.""" + login_details = self._get_or_create_admin_account() + if not login_details or not login_details.token: + logger.warning("couldn't distribute certificates: not logged in") + return + databag_csrs = self.tls.get_certificate_requests() + notary_table = self.client.get_certificate_requests_table(login_details.token) + if not notary_table: + logger.warning("couldn't distribute certificates: couldn't get table from notary") + return + + for request in databag_csrs: + notary_rows_with_matching_csr = [ + row + for row in notary_table.rows + if row.csr == str(request.certificate_signing_request) + ] + if len(notary_rows_with_matching_csr) < 1: + self.client.post_csr(str(request.certificate_signing_request), login_details.token) + continue + assert len(notary_rows_with_matching_csr) < 2 + request_notary_entry = notary_rows_with_matching_csr[0] + certificates_provided_for_csr = [ + csr + for csr in self.tls.get_issued_certificates(request.relation_id) + if str(csr.certificate_signing_request) == request_notary_entry.csr + ] + if ( + request_notary_entry.certificate_chain == "rejected" + or request_notary_entry.certificate_chain == "" + ): + if len(certificates_provided_for_csr) > 0: + last_provided_certificate = certificates_provided_for_csr[0] + self.tls.set_relation_certificate( + ProviderCertificate( + relation_id=request.relation_id, + certificate_signing_request=request.certificate_signing_request, + certificate=last_provided_certificate.certificate, + ca=last_provided_certificate.ca, + chain=last_provided_certificate.chain, + revoked=True, + ) + ) + continue + certificate_chain = [ + Certificate.from_string(cert) for cert in request_notary_entry.certificate_chain + ] + certificate_not_provided_yet = ( + len(certificate_chain) > 0 and len(certificates_provided_for_csr) == 0 + ) + certificate_provided_is_stale = ( + len(certificate_chain) > 0 + and len(certificates_provided_for_csr) == 1 + and certificate_chain[0] != certificates_provided_for_csr[0].certificate + ) + if certificate_not_provided_yet or certificate_provided_is_stale: + self.tls.set_relation_certificate( + ProviderCertificate( + relation_id=request.relation_id, + certificate_signing_request=request.certificate_signing_request, + certificate=certificate_chain[0], + ca=certificate_chain[-1], + chain=certificate_chain, + ) + ) + ## Properties ## @property def _pebble_layer(self) -> ops.pebble.LayerDict: """Return a dictionary representing a Pebble layer.""" return { - "summary": "gocert layer", - "description": "pebble config layer for gocert", + "summary": "notary layer", + "description": "pebble config layer for notary", "services": { - "gocert": { + "notary": { "override": "replace", - "summary": "gocert", - "command": f"gocert -config {WORKLOAD_CONFIG_PATH}/config/config.yaml", + "summary": "notary", + "command": f"notary -config {WORKLOAD_CONFIG_PATH}/config/config.yaml", "startup": "enabled", } }, @@ -267,10 +350,10 @@ def _get_or_create_admin_account(self) -> LoginSecret | None: """Get the first admin user for the charm to use from secrets. Create one if it doesn't exist. Returns: - Login details secret if they exist. None if the related account couldn't be created in GoCert. + Login details secret if they exist. None if the related account couldn't be created in Notary. """ try: - secret = self.model.get_secret(label=GOCERT_LOGIN_SECRET_LABEL) + secret = self.model.get_secret(label=NOTARY_LOGIN_SECRET_LABEL) secret_content = secret.get_content(refresh=True) username = secret_content.get("username", "") password = secret_content.get("password", "") @@ -281,7 +364,7 @@ def _get_or_create_admin_account(self) -> LoginSecret | None: password = _generate_password() account = LoginSecret(username, password, None) self.app.add_secret( - label=GOCERT_LOGIN_SECRET_LABEL, + label=NOTARY_LOGIN_SECRET_LABEL, content=account.to_dict(), ) logger.info("admin account details saved to secrets.") @@ -293,7 +376,7 @@ def _get_or_create_admin_account(self) -> LoginSecret | None: def _generate_password() -> str: - """Generate a password for the GoCert Account.""" + """Generate a password for the Notary Account.""" pw = [] pw.append(random.choice(string.ascii_lowercase)) pw.append(random.choice(string.ascii_uppercase)) @@ -306,10 +389,10 @@ def _generate_password() -> str: def _generate_username() -> str: - """Generate a username for the GoCert Account.""" + """Generate a username for the Notary Account.""" suffix = [random.choice(string.ascii_uppercase) for i in range(4)] return "charm-admin-" + "".join(suffix) if __name__ == "__main__": # pragma: nocover - ops.main(GocertCharm) # type: ignore + ops.main(NotaryCharm) # type: ignore diff --git a/src/config/config.yaml b/src/config/config.yaml deleted file mode 100644 index 000eace..0000000 --- a/src/config/config.yaml +++ /dev/null @@ -1,5 +0,0 @@ -key_path: "/etc/gocert/config/private_key.pem" -cert_path: "/etc/gocert/config/certificate.pem" -db_path: "/var/lib/gocert/database/certs.db" -port: 2111 -pebble_notifications: true \ No newline at end of file diff --git a/src/gocert.py b/src/gocert.py deleted file mode 100644 index ae09f04..0000000 --- a/src/gocert.py +++ /dev/null @@ -1,115 +0,0 @@ -# Copyright 2024 Canonical Ltd. -# See LICENSE file for licensing details. - -"""Library for interacting with the GoCert application.""" - -import logging - -import requests - -logger = logging.getLogger(__name__) - - -class GoCertClientError(Exception): - """Base class for exceptions raised by the GoCert client.""" - - -class GoCert: - """Class to interact with GoCert.""" - - API_VERSION = "v1" - - def __init__(self, url: str, ca_path: str) -> None: - """Initialize a client for interacting with GoCert. - - Args: - url: the endpoint that gocert is listening on e.g https://gocert.com:8000 - ca_path: the file path that contains the ca cert that gocert uses for https communication - """ - self.url = url - self.ca_path = ca_path - - def login(self, username: str, password: str) -> str | None: - """Login to gocert by sending the username and password and return a Token.""" - try: - req = requests.post( - f"{self.url}/login", - verify=self.ca_path if self.ca_path else None, - json={"username": username, "password": password}, - ) - except (requests.RequestException, OSError): - return - try: - req.raise_for_status() - except requests.HTTPError: - logger.error("couldn't log in: code %s, %s", req.status_code, req.text) - return - logger.info("logged in to GoCert successfully") - return req.text - - def token_is_valid(self, token: str) -> bool: - """Return if the token is still valid by attempting to connect to an endpoint.""" - try: - req = requests.get( - f"{self.url}/accounts", - verify=self.ca_path if self.ca_path else None, - headers={"Authorization": f"Bearer {token}"}, - ) - req.raise_for_status() - except (requests.RequestException, OSError): - return False - return True - - def is_api_available(self) -> bool: - """Return if the GoCert server is reachable.""" - try: - req = requests.get( - f"{self.url}/status", - verify=self.ca_path if self.ca_path else None, - ) - req.raise_for_status() - except (requests.RequestException, OSError): - return False - return True - - def is_initialized(self) -> bool: - """Return if the GoCert server is initialized.""" - try: - req = requests.get( - f"{self.url}/status", - verify=self.ca_path if self.ca_path else None, - ) - req.raise_for_status() - except (requests.RequestException, OSError): - return False - body = req.json() - return body.get("initialized", False) - - def create_first_user(self, username: str, password: str) -> int | None: - """Create the first admin user. - - Args: - username: username of the first user - password: password for the first user. It must be longer than 7 characters, have at least one lowercase, - one uppercase and one number or special character. - - Returns: - int | None: the id of the created user, or None if the request failed - - """ - try: - req = requests.post( - f"{self.url}/api/{self.API_VERSION}/accounts", - verify=self.ca_path if self.ca_path else None, - json={"username": username, "password": password}, - ) - except (requests.RequestException, OSError): - return None - try: - req.raise_for_status() - except requests.HTTPError: - logger.warning("couldn't create first user: code %s, %s", req.status_code, req.text) - return None - logger.info("created the first user in GoCert.") - id = req.json().get("id") - return int(id) if id else None diff --git a/src/grafana_dashboards/gocert.json b/src/grafana_dashboards/notary.json similarity index 98% rename from src/grafana_dashboards/gocert.json rename to src/grafana_dashboards/notary.json index 5d4dabf..9db73ab 100644 --- a/src/grafana_dashboards/gocert.json +++ b/src/grafana_dashboards/notary.json @@ -39,7 +39,7 @@ "type": "prometheus", "uid": "${prometheusds}" }, - "description": "The total number of certificate requests in GoCert", + "description": "The total number of certificate requests in Notary", "fieldConfig": { "defaults": { "color": { @@ -443,7 +443,7 @@ "uid": "${prometheusds}" }, "editorMode": "builder", - "expr": "go_goroutines{juju_application=~\"$juju_application\",juju_charm=\"gocert-k8s\",juju_model=~\"$juju_model\",juju_model_uuid=~\"$juju_model_uuid\",juju_unit=~\"$juju_unit\"}", + "expr": "go_goroutines{juju_application=~\"$juju_application\",juju_charm=\"notary-k8s\",juju_model=~\"$juju_model\",juju_model_uuid=~\"$juju_model_uuid\",juju_unit=~\"$juju_unit\"}", "legendFormat": "Goroutines", "range": true, "refId": "A" @@ -457,7 +457,7 @@ "type": "prometheus", "uid": "${prometheusds}" }, - "description": "The total memory usage by GoCert", + "description": "The total memory usage by Notary", "fieldConfig": { "defaults": { "color": { @@ -506,7 +506,7 @@ "uid": "${prometheusds}" }, "editorMode": "builder", - "expr": "go_memstats_alloc_bytes{juju_charm=\"gocert-k8s\"}", + "expr": "go_memstats_alloc_bytes{juju_charm=\"notary-k8s\"}", "legendFormat": "__auto", "range": true, "refId": "A" @@ -557,12 +557,12 @@ "uid": "${lokids}" }, "editorMode": "builder", - "expr": "{charm=\"gocert-k8s\", juju_application=~\"$juju_application\", juju_model=~\"$juju_model\", juju_model_uuid=~\"$juju_model_uuid\", juju_unit=~\"$juju_unit\"} |= \"\"", + "expr": "{charm=\"notary-k8s\", juju_application=~\"$juju_application\", juju_model=~\"$juju_model\", juju_model_uuid=~\"$juju_model_uuid\", juju_unit=~\"$juju_unit\"} |= \"\"", "queryType": "range", "refId": "A" } ], - "title": "GoCert Logs", + "title": "Notary Logs", "type": "logs" } ], @@ -786,7 +786,7 @@ }, "timepicker": {}, "timezone": "", - "title": "GoCert", + "title": "Notary", "uid": "1546e4c5ed06bcba", "version": 1, "weekStart": "" diff --git a/src/notary.py b/src/notary.py new file mode 100644 index 0000000..c7f92f9 --- /dev/null +++ b/src/notary.py @@ -0,0 +1,230 @@ +# Copyright 2024 Canonical Ltd. +# See LICENSE file for licensing details. + +"""Library for interacting with the Notary application.""" + +import logging +from dataclasses import dataclass +from typing import Literal + +import requests + +logger = logging.getLogger(__name__) + + +class NotaryClientError(Exception): + """Base class for exceptions raised by the Notary client.""" + + +@dataclass(frozen=True) +class CertificateRequest: + """The certificate request that's stored in Notary.""" + + id: int + csr: str + certificate_chain: list[str] | Literal["", "rejected"] + + +@dataclass +class CertificateRequests: + """The table of certificate requests in Notary.""" + + rows: list[CertificateRequest] + + +class Notary: + """Class to interact with Notary.""" + + API_VERSION = "v1" + + def __init__(self, url: str, ca_path: str | bool = False) -> None: + """Initialize a client for interacting with Notary. + + Args: + url: the endpoint that notary is listening on e.g https://notary.com:8000 + ca_path: the file path that contains the ca cert that notary uses for https communication + """ + self.url = url + self.ca_path = ca_path + + def login(self, username: str, password: str) -> str | None: + """Login to notary by sending the username and password and return a Token.""" + try: + req = requests.post( + f"{self.url}/login", + verify=self.ca_path, + json={"username": username, "password": password}, + ) + except (requests.RequestException, OSError): + return + try: + req.raise_for_status() + except requests.HTTPError: + logger.error("couldn't log in: code %s, %s", req.status_code, req.text) + return + logger.info("logged in to Notary successfully") + return req.text + + def token_is_valid(self, token: str) -> bool: + """Return if the token is still valid by attempting to connect to an endpoint.""" + try: + req = requests.get( + f"{self.url}/api/{self.API_VERSION}/accounts/me", + verify=self.ca_path, + headers={"Authorization": f"Bearer {token}"}, + ) + req.raise_for_status() + except (requests.RequestException, OSError): + return False + return True + + def is_api_available(self) -> bool: + """Return if the Notary server is reachable.""" + try: + req = requests.get( + f"{self.url}/status", + verify=self.ca_path, + ) + req.raise_for_status() + except (requests.RequestException, OSError): + return False + return True + + def is_initialized(self) -> bool: + """Return if the Notary server is initialized.""" + try: + req = requests.get( + f"{self.url}/status", + verify=self.ca_path, + ) + req.raise_for_status() + except (requests.RequestException, OSError): + return False + body = req.json() + return body.get("initialized", False) + + def create_first_user(self, username: str, password: str) -> int | None: + """Create the first admin user. + + Args: + username: username of the first user + password: password for the first user. It must be longer than 7 characters, have at least one lowercase, + one uppercase and one number or special character. + + Returns: + int | None: the id of the created user, or None if the request failed + + """ + try: + req = requests.post( + f"{self.url}/api/{self.API_VERSION}/accounts", + verify=self.ca_path, + json={"username": username, "password": password}, + ) + except (requests.RequestException, OSError): + return None + try: + req.raise_for_status() + except requests.HTTPError: + logger.error("couldn't create first user: code %s, %s", req.status_code, req.text) + return None + logger.info("created the first user in Notary.") + id = req.json().get("id") + return int(id) if id else None + + def get_certificate_requests_table(self, token: str) -> CertificateRequests | None: + """Get all certificate requests table from Notary. + + Returns: + None if the request fails to go through. The table itself, otherwise. + """ + try: + res = requests.get( + f"{self.url}/api/{self.API_VERSION}/certificate_requests", + verify=self.ca_path, + headers={"Authorization": f"Bearer {token}"}, + ) + res.raise_for_status() + except requests.RequestException as e: + logger.error( + "couldn't retrieve certificate requests table: code %s, %s", + e.response.status_code if e.response else "unknown", + e.response.text if e.response else "unknown", + ) + return None + except OSError: + logger.error("error occurred during HTTP request: TLS file invalid") + return None + table = res.json() + return CertificateRequests( + rows=[ + CertificateRequest( + row.get("id"), + row.get("csr"), + serialize(row.get("certificate")), + ) + for row in table + ] + if table + else [] + ) + + def post_csr(self, csr: str, token: str) -> None: + """Post a new CSR to Notary.""" + try: + res = requests.post( + f"{self.url}/api/{self.API_VERSION}/certificate_requests", + verify=self.ca_path, + headers={"Authorization": f"Bearer {token}"}, + data=csr, + ) + res.raise_for_status() + except requests.RequestException as e: + logger.error( + "couldn't post new certificate requests: code %s, %s", + e.response.status_code if e.response else "unknown", + e.response.text if e.response else "unknown", + ) + except OSError: + logger.error("error occurred during HTTP request: TLS file invalid") + + def post_certificate(self, csr: str, cert_chain: list[str], token: str) -> None: + """Post a certificate chain to an associated csr to Notary.""" + try: + table = self.get_certificate_requests_table(token) + if not table: + return + csr_ids = list(filter(lambda x: x.csr == csr, table.rows)) + if len(csr_ids) != 1: + logger.error("given CSR not found in Notary") + return + res = requests.post( + f"{self.url}/api/{self.API_VERSION}/certificate_requests/{csr_ids[0].id}/certificate", + verify=self.ca_path, + headers={"Authorization": f"Bearer {token}"}, + data="\n".join(cert_chain), + ) + res.raise_for_status() + except requests.RequestException as e: + logger.error( + "couldn't post new certificate: code %s, %s", + e.response.status_code if e.response else "unknown", + e.response.text if e.response else "unknown", + ) + except OSError: + logger.error("error occurred during HTTP request: TLS file invalid") + + +def serialize(pem_string: str) -> list[str] | Literal["", "rejected"]: + """Process the certificate entry coming from Notary. + + Returns: + a list of pem strings, an empty string or a rejected string. + """ + if pem_string != "" and pem_string != "rejected": + return [ + cert.strip() + "-----END CERTIFICATE-----" + for cert in pem_string.split("-----END CERTIFICATE-----") + if cert.strip() + ] + return pem_string diff --git a/test-requirements.txt b/test-requirements.txt index a04131d..2e020e7 100644 --- a/test-requirements.txt +++ b/test-requirements.txt @@ -24,7 +24,7 @@ codespell==2.3.0 # via -r test-requirements.in coverage[toml]==7.6.1 # via -r test-requirements.in -cryptography==43.0.0 +cryptography==43.0.1 # via paramiko decorator==5.1.1 # via @@ -68,9 +68,9 @@ oauthlib==3.2.2 # via # kubernetes # requests-oauthlib -ops==2.15.0 +ops==2.16.1 # via ops-scenario -ops-scenario==6.1.5 +ops-scenario==7.0.2 # via -r test-requirements.in packaging==24.1 # via @@ -114,9 +114,9 @@ pyrfc3339==1.1 # via # juju # macaroonbakery -pyright==1.1.377 +pyright==1.1.380 # via -r test-requirements.in -pytest==8.3.2 +pytest==8.3.3 # via # -r test-requirements.in # pytest-asyncio @@ -146,7 +146,7 @@ requests-oauthlib==2.0.0 # via kubernetes rsa==4.9 # via google-auth -ruff==0.6.2 +ruff==0.6.5 # via -r test-requirements.in six==1.16.0 # via diff --git a/tests/integration/test_charm.py b/tests/integration/test_charm.py index fc79874..25f023e 100644 --- a/tests/integration/test_charm.py +++ b/tests/integration/test_charm.py @@ -3,13 +3,25 @@ # See LICENSE file for licensing details. import asyncio +import json import logging +from base64 import b64decode from pathlib import Path import pytest import yaml +from charms.tls_certificates_interface.v4.tls_certificates import ( + CertificateSigningRequest, + generate_ca, + generate_certificate, + generate_private_key, +) +from juju.client.client import SecretsFilter from pytest_operator.plugin import OpsTest +from charm import NOTARY_LOGIN_SECRET_LABEL +from notary import Notary + logger = logging.getLogger(__name__) CHARMCRAFT = yaml.safe_load(Path("./charmcraft.yaml").read_text()) @@ -18,29 +30,85 @@ LOKI_APPLICATION_NAME = "loki-k8s" PROMETHEUS_APPLICATION_NAME = "prometheus-k8s" TRAEIK_K8S_APPLICATION_NAME = "traefik-k8s" +TLS_REQUIRER_APPLICATION_NAME = "tls-certificates-requirer" @pytest.mark.abort_on_fail -async def test_build_and_deploy(ops_test: OpsTest, request): +async def test_build_and_deploy(ops_test: OpsTest, request: pytest.FixtureRequest): """Build the charm-under-test and deploy it together with related charms. Assert on the unit status before any relations/configurations take place. """ - charm = Path(request.config.getoption("--charm_path")).resolve() - resources = {"gocert-image": CHARMCRAFT["resources"]["gocert-image"]["upstream-source"]} + charm = Path(request.config.getoption("--charm_path")).resolve() # type: ignore + resources = {"notary-image": CHARMCRAFT["resources"]["notary-image"]["upstream-source"]} - # Deploy the charm and wait for active status - await asyncio.gather( - ops_test.model.deploy(charm, resources=resources, application_name=APP_NAME), - ops_test.model.wait_for_idle(apps=[APP_NAME], status="active", timeout=1000), + assert ops_test.model + await ops_test.model.deploy(charm, resources=resources, application_name=APP_NAME) + await ops_test.model.wait_for_idle(apps=[APP_NAME], status="active", timeout=1000) + + +@pytest.mark.abort_on_fail +async def test_given_notary_when_tls_requirer_related_then_csr_uploaded_to_notary_and_certificate_provided_to_requirer( + ops_test: OpsTest, +): + assert ops_test.model + admin_credentials = await get_notary_credentials(ops_test) + token = admin_credentials.get("token") + assert token + endpoint = await get_notary_endpoint(ops_test) + client = Notary(url=endpoint) + assert client.token_is_valid(token) + + await ops_test.model.deploy( + "tls-certificates-requirer", + application_name=TLS_REQUIRER_APPLICATION_NAME, + channel="edge", + trust=True, + ) + await ops_test.model.integrate( + relation1=f"{APP_NAME}:certificates", + relation2=f"{TLS_REQUIRER_APPLICATION_NAME}", + ) + await ops_test.model.wait_for_idle( + apps=[APP_NAME, TLS_REQUIRER_APPLICATION_NAME], + status="active", + timeout=1000, + raise_on_error=True, + ) + table = client.get_certificate_requests_table(token) + assert table + assert len(table.rows) == 1 + + row = table.rows[0] + ca_pk = generate_private_key() + ca = generate_ca(ca_pk, 365, "integration-test") + cert = generate_certificate(CertificateSigningRequest.from_string(row.csr), ca, ca_pk, 365) + chain = [str(cert), str(ca)] + client.post_certificate(row.csr, chain, token) + + table = client.get_certificate_requests_table(token) + assert table + assert table.rows[0].certificate_chain != "" + assert table.rows[0].certificate_chain != "rejected" + + await ops_test.model.wait_for_idle( + apps=[APP_NAME, TLS_REQUIRER_APPLICATION_NAME], + status="active", + timeout=1000, + raise_on_error=True, ) + action_result = await run_get_certificate_action(ops_test) + given_certificate: str = json.loads(action_result)[0].get("certificate", "") + assert given_certificate.replace("\n", "") == str(cert).replace("\n", "") + @pytest.mark.abort_on_fail -async def test_given_loki_and_prometheus_related_to_gocert_all_charm_statuses_active( +async def test_given_loki_and_prometheus_related_to_notary_all_charm_statuses_active( ops_test: OpsTest, ): """Deploy loki and prometheus, and make sure all applications are active.""" + assert ops_test.model deploy_prometheus = ops_test.model.deploy( "prometheus-k8s", application_name=PROMETHEUS_APPLICATION_NAME, @@ -85,3 +153,38 @@ async def test_given_application_deployed_when_related_to_traefik_k8s_then_all_s timeout=1000, raise_on_error=True, ) +async def get_notary_endpoint(ops_test: OpsTest) -> str: + assert ops_test.model + status = await ops_test.model.get_status() + notary_ip = status.applications[APP_NAME].units[f"{APP_NAME}/0"].address + return f"https://{notary_ip}:2111" + + +async def get_notary_credentials(ops_test: OpsTest) -> dict[str, str]: + assert ops_test.model + secrets = await ops_test.model.list_secrets( + filter=SecretsFilter(label=NOTARY_LOGIN_SECRET_LABEL), show_secrets=True + ) + return { + field: b64decode(secrets[0].value.data[field]).decode("utf-8") + for field in ["username", "password", "token"] + } + + +async def run_get_certificate_action(ops_test: OpsTest) -> str: + """Run `get-certificate` on the `tls-requirer-requirer/0` unit. + + Args: + ops_test (OpsTest): OpsTest + + Returns: + dict: Action output + """ + assert ops_test.model + tls_requirer_unit = ops_test.model.units[f"{TLS_REQUIRER_APPLICATION_NAME}/0"] + assert tls_requirer_unit + action = await tls_requirer_unit.run_action( + action_name="get-certificate", + ) + action_output = await ops_test.model.get_action_output(action_uuid=action.entity_id, wait=30) + return action_output.get("certificates", "") diff --git a/tests/unit/test_charm.py b/tests/unit/test_charm.py index a32f632..c004de2 100644 --- a/tests/unit/test_charm.py +++ b/tests/unit/test_charm.py @@ -5,28 +5,33 @@ from unittest.mock import Mock, patch import ops -import ops.testing import pytest -from scenario import Container, Context, Event, Mount, Network, State, Storage +from ops.pebble import Layer +from scenario import Container, Context, Mount, Network, Relation, Secret, State, Storage -from charm import GocertCharm +from charm import CERTIFICATE_PROVIDER_RELATION_NAME, NOTARY_LOGIN_SECRET_LABEL, NotaryCharm from lib.charms.tls_certificates_interface.v4.tls_certificates import ( Certificate, PrivateKey, + ProviderCertificate, + RequirerCSR, generate_ca, generate_certificate, generate_csr, generate_private_key, ) +from notary import CertificateRequest, CertificateRequests -CERTIFICATE_COMMON_NAME = "GoCert Self Signed Certificate" -SELF_SIGNED_CA_COMMON_NAME = "GoCert Self Signed Root CA" +TLS_LIB_PATH = "charms.tls_certificates_interface.v4.tls_certificates" + +CERTIFICATE_COMMON_NAME = "Notary Self Signed Certificate" +SELF_SIGNED_CA_COMMON_NAME = "Notary Self Signed Root CA" class TestCharm: @pytest.fixture(scope="function") def context(self): - yield Context(GocertCharm) + yield Context(NotaryCharm) def example_cert_and_key(self) -> tuple[Certificate, PrivateKey]: private_key = generate_private_key() @@ -49,1385 +54,2897 @@ def example_cert_and_key(self) -> tuple[Certificate, PrivateKey]: return certificate, private_key # Configure tests - def test_given_only_config_storage_container_cant_connect_network_not_available_gocert_not_running_when_configure_then_no_error_raised( - self, context - ): - state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network([], [], [])}, + def test_given_only_config_storage_container_cant_connect_network_not_available_notary_not_running_when_configure_then_no_error_raised( + self, context + ): + state = State( + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_only_database_storage_container_cant_connect_network_not_available_gocert_not_running_when_configure_then_no_error_raised( + def test_given_only_database_storage_container_cant_connect_network_not_available_notary_not_running_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network([], [], [])}, + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_storages_available_container_cant_connect_network_not_available_gocert_not_running_when_configure_then_no_error_raised( + def test_given_storages_available_container_cant_connect_network_not_available_notary_not_running_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network([], [], [])}, + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_only_config_storage_container_can_connect_network_not_available_gocert_not_running_when_configure_then_no_error_raised( + def test_given_only_config_storage_container_can_connect_network_not_available_notary_not_running_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network([], [], [])}, + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_only_database_storage_container_can_connect_network_not_available_gocert_not_running_when_configure_then_no_error_raised( + def test_given_only_database_storage_container_can_connect_network_not_available_notary_not_running_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network([], [], [])}, + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_storages_available_container_can_connect_network_not_available_gocert_not_running_when_configure_then_config_file_generated( + def test_given_storages_available_container_can_connect_network_not_available_notary_not_running_when_configure_then_config_file_generated( self, context ): state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network([], [], [])}, + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("config-changed"), state) - root = out.containers[0].get_filesystem(context) - assert (root / "etc/gocert/config/config.yaml").open("r") - assert not (root / "etc/gocert/config/certificate.pem").exists() - assert not ((root / "etc/gocert/config/private_key.pem").exists()) + out = context.run(context.on.config_changed(), state) + root = out.get_container("notary").get_filesystem(context) + assert (root / "etc/notary/config/config.yaml").open("r") + assert not (root / "etc/notary/config/certificate.pem").exists() + assert not (root / "etc/notary/config/private_key.pem").exists() assert len(out.secrets) == 1 - assert out.secrets[0].label == "GoCert Login Details" - - def test_given_only_config_storage_container_cant_connect_network_available_gocert_not_running_when_configure_then_no_error_raised( - self, context - ): - state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network.default()}, + assert out.get_secret(label="Notary Login Details") + + def test_given_only_config_storage_container_cant_connect_network_available_notary_not_running_when_configure_then_no_error_raised( + self, context + ): + state = State( + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_only_database_storage_container_cant_connect_network_available_gocert_not_running_when_configure_then_no_error_raised( + def test_given_only_database_storage_container_cant_connect_network_available_notary_not_running_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network.default()}, + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_storages_available_container_cant_connect_network_available_gocert_not_running_when_configure_then_no_error_raised( + def test_given_storages_available_container_cant_connect_network_available_notary_not_running_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network.default()}, + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_only_config_storage_container_can_connect_network_available_gocert_not_running_when_configure_then_no_error_raised( + def test_given_only_config_storage_container_can_connect_network_available_notary_not_running_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network.default()}, + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_only_database_storage_container_can_connect_network_available_gocert_not_running_when_configure_then_no_error_raised( + def test_given_only_database_storage_container_can_connect_network_available_notary_not_running_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network.default()}, + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_storages_available_container_can_connect_network_available_gocert_not_running_when_configure_then_config_and_certificates_generated( + def test_given_storages_available_container_can_connect_network_available_notary_not_running_when_configure_then_config_and_certificates_generated( self, context ): state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network.default()}, + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("config-changed"), state) - root = out.containers[0].get_filesystem(context) - assert (root / "etc/gocert/config/config.yaml").open("r") + out = context.run(context.on.config_changed(), state) + root = out.get_container("notary").get_filesystem(context) + assert (root / "etc/notary/config/config.yaml").open("r") assert ( - (root / "etc/gocert/config/certificate.pem") + (root / "etc/notary/config/certificate.pem") .open("r") .read() .startswith("-----BEGIN CERTIFICATE-----") ) assert ( - (root / "etc/gocert/config/private_key.pem") + (root / "etc/notary/config/private_key.pem") .open("r") .read() .startswith("-----BEGIN RSA PRIVATE KEY-----") ) - def test_given_only_config_storage_container_cant_connect_network_not_available_gocert_running_when_configure_then_no_error_raised( - self, context - ): - state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network([], [], [])}, + def test_given_only_config_storage_container_cant_connect_network_not_available_notary_running_when_configure_then_no_error_raised( + self, context + ): + state = State( + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_only_database_storage_container_cant_connect_network_not_available_gocert_running_when_configure_then_no_error_raised( + def test_given_only_database_storage_container_cant_connect_network_not_available_notary_running_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network([], [], [])}, + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_storages_available_container_cant_connect_network_not_available_gocert_running_when_configure_then_no_error_raised( + def test_given_storages_available_container_cant_connect_network_not_available_notary_running_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network([], [], [])}, + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_only_config_storage_container_can_connect_network_not_available_gocert_running_when_configure_then_no_error_raised( + def test_given_only_config_storage_container_can_connect_network_not_available_notary_running_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network([], [], [])}, + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_only_database_storage_container_can_connect_network_not_available_gocert_running_when_configure_then_no_error_raised( + def test_given_only_database_storage_container_can_connect_network_not_available_notary_running_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network([], [], [])}, + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_storages_available_container_can_connect_network_not_available_gocert_running_when_configure_then_config_file_generated( + def test_given_storages_available_container_can_connect_network_not_available_notary_running_when_configure_then_config_file_generated( self, context ): state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network([], [], [])}, + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("config-changed"), state) - root = out.containers[0].get_filesystem(context) - assert (root / "etc/gocert/config/config.yaml").open("r") - assert not (root / "etc/gocert/config/certificate.pem").exists() - assert not ((root / "etc/gocert/config/private_key.pem").exists()) + out = context.run(context.on.config_changed(), state) + root = out.get_container("notary").get_filesystem(context) + assert (root / "etc/notary/config/config.yaml").open("r") + assert not (root / "etc/notary/config/certificate.pem").exists() + assert not ((root / "etc/notary/config/private_key.pem").exists()) assert len(out.secrets) == 1 - assert out.secrets[0].label == "GoCert Login Details" - - def test_given_only_config_storage_container_cant_connect_network_available_gocert_running_when_configure_then_no_error_raised( - self, context - ): - state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network.default()}, + assert out.get_secret(label="Notary Login Details") + + def test_given_only_config_storage_container_cant_connect_network_available_notary_running_when_configure_then_no_error_raised( + self, context + ): + state = State( + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_only_database_storage_container_cant_connect_network_available_gocert_running_when_configure_then_no_error_raised( + def test_given_only_database_storage_container_cant_connect_network_available_notary_running_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network.default()}, + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_storages_available_container_cant_connect_network_available_gocert_running_when_configure_then_no_error_raised( + def test_given_storages_available_container_cant_connect_network_available_notary_running_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network.default()}, + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_only_config_storage_container_can_connect_network_available_gocert_running_when_configure_then_no_error_raised( + def test_given_only_config_storage_container_can_connect_network_available_notary_running_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network.default()}, + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_only_database_storage_container_can_connect_network_available_gocert_running_when_configure_then_no_error_raised( + def test_given_only_database_storage_container_can_connect_network_available_notary_running_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network.default()}, + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_storages_available_container_can_connect_network_available_gocert_running_when_configure_then_status_is_blocked( + def test_given_storages_available_container_can_connect_network_available_notary_running_when_configure_then_status_is_blocked( self, context ): state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network.default()}, + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_only_config_storage_container_cant_connect_network_not_available_gocert_initialized_when_configure_then_no_error_raised( + def test_given_only_config_storage_container_cant_connect_network_not_available_notary_initialized_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network([], [], [])}, + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_only_database_storage_container_cant_connect_network_not_available_gocert_initialized_when_configure_then_no_error_raised( + def test_given_only_database_storage_container_cant_connect_network_not_available_notary_initialized_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network([], [], [])}, + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_storages_available_container_cant_connect_network_not_available_gocert_initialized_when_configure_then_no_error_raised( + def test_given_storages_available_container_cant_connect_network_not_available_notary_initialized_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network([], [], [])}, + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_only_config_storage_container_can_connect_network_not_available_gocert_initialized_when_configure_then_no_error_raised( + def test_given_only_config_storage_container_can_connect_network_not_available_notary_initialized_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network([], [], [])}, + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_only_database_storage_container_can_connect_network_not_available_gocert_initialized_when_configure_then_no_error_raised( + def test_given_only_database_storage_container_can_connect_network_not_available_notary_initialized_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network([], [], [])}, + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_storages_available_container_can_connect_network_not_available_gocert_initialized_when_configure_then_config_file_generated( + def test_given_storages_available_container_can_connect_network_not_available_notary_initialized_when_configure_then_config_file_generated( self, context ): state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network([], [], [])}, + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - out = context.run(Event("config-changed"), state) + out = context.run(context.on.config_changed(), state) - root = out.containers[0].get_filesystem(context) - assert (root / "etc/gocert/config/config.yaml").open("r") - assert not (root / "etc/gocert/config/certificate.pem").exists() - assert not ((root / "etc/gocert/config/private_key.pem").exists()) + root = out.get_container("notary").get_filesystem(context) + assert (root / "etc/notary/config/config.yaml").open("r") + assert not (root / "etc/notary/config/certificate.pem").exists() + assert not ((root / "etc/notary/config/private_key.pem").exists()) assert len(out.secrets) == 1 - assert out.secrets[0].label == "GoCert Login Details" - - def test_given_only_config_storage_container_cant_connect_network_available_gocert_initialized_when_configure_then_no_error_raised( - self, context - ): - state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network.default()}, + assert out.get_secret(label="Notary Login Details") + + def test_given_only_config_storage_container_cant_connect_network_available_notary_initialized_when_configure_then_no_error_raised( + self, context + ): + state = State( + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_only_database_storage_container_cant_connect_network_available_gocert_initialized_when_configure_then_no_error_raised( + def test_given_only_database_storage_container_cant_connect_network_available_notary_initialized_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network.default()}, + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_storages_available_container_cant_connect_network_available_gocert_initialized_when_configure_then_no_error_raised( + def test_given_storages_available_container_cant_connect_network_available_notary_initialized_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network.default()}, + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_only_config_storage_container_can_connect_network_available_gocert_initialized_when_configure_then_no_error_raised( + def test_given_only_config_storage_container_can_connect_network_available_notary_initialized_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network.default()}, + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_only_database_storage_container_can_connect_network_available_gocert_initialized_when_configure_then_no_error_raised( + def test_given_only_database_storage_container_can_connect_network_available_notary_initialized_when_configure_then_no_error_raised( self, context ): state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network.default()}, + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) - def test_given_storages_available_container_can_connect_network_available_gocert_initialized_when_configure_then_status_is_active( + def test_given_storages_available_container_can_connect_network_available_notary_initialized_when_configure_then_status_is_active( self, context ): state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network.default()}, + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - context.run(Event("config-changed"), state) + context.run(context.on.config_changed(), state) # Unit Status Tests - def test_given_only_config_storage_container_cant_connect_network_not_available_gocert_not_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network([], [], [])}, + def test_given_only_config_storage_container_cant_connect_network_not_available_notary_not_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("container not yet connectable") - def test_given_only_database_storage_container_cant_connect_network_not_available_gocert_not_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network([], [], [])}, + def test_given_only_database_storage_container_cant_connect_network_not_available_notary_not_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("container not yet connectable") - def test_given_storages_available_container_cant_connect_network_not_available_gocert_not_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network([], [], [])}, + def test_given_storages_available_container_cant_connect_network_not_available_notary_not_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("container not yet connectable") - def test_given_only_config_storage_container_can_connect_network_not_available_gocert_not_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network([], [], [])}, + def test_given_only_config_storage_container_can_connect_network_not_available_notary_not_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("storages not yet available") - def test_given_only_database_storage_container_can_connect_network_not_available_gocert_not_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network([], [], [])}, + def test_given_only_database_storage_container_can_connect_network_not_available_notary_not_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("storages not yet available") - def test_given_storages_available_container_can_connect_network_not_available_gocert_not_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network([], [], [])}, + def test_given_storages_available_container_can_connect_network_not_available_notary_not_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("certificates not yet created") - def test_given_only_config_storage_container_cant_connect_network_available_gocert_not_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network.default()}, + def test_given_only_config_storage_container_cant_connect_network_available_notary_not_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("container not yet connectable") - def test_given_only_database_storage_container_cant_connect_network_available_gocert_not_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network.default()}, + def test_given_only_database_storage_container_cant_connect_network_available_notary_not_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("container not yet connectable") - def test_given_storages_available_container_cant_connect_network_available_gocert_not_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network.default()}, + def test_given_storages_available_container_cant_connect_network_available_notary_not_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("container not yet connectable") - def test_given_only_config_storage_container_can_connect_network_available_gocert_not_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network.default()}, + def test_given_only_config_storage_container_can_connect_network_available_notary_not_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("storages not yet available") - def test_given_only_database_storage_container_can_connect_network_available_gocert_not_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network.default()}, + def test_given_only_database_storage_container_can_connect_network_available_notary_not_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("storages not yet available") - def test_given_storages_available_container_can_connect_network_available_gocert_not_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network.default()}, + def test_given_storages_available_container_can_connect_network_available_notary_not_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": False, "is_initialized.return_value": False}, + **{"is_api_available.return_value": False, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("certificates not yet created") - def test_given_only_config_storage_container_cant_connect_network_not_available_gocert_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network([], [], [])}, + def test_given_only_config_storage_container_cant_connect_network_not_available_notary_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("container not yet connectable") - def test_given_only_database_storage_container_cant_connect_network_not_available_gocert_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network([], [], [])}, + def test_given_only_database_storage_container_cant_connect_network_not_available_notary_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("container not yet connectable") - def test_given_storages_available_container_cant_connect_network_not_available_gocert_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network([], [], [])}, + def test_given_storages_available_container_cant_connect_network_not_available_notary_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("container not yet connectable") - def test_given_only_config_storage_container_can_connect_network_not_available_gocert_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network([], [], [])}, + def test_given_only_config_storage_container_can_connect_network_not_available_notary_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("storages not yet available") - def test_given_only_database_storage_container_can_connect_network_not_available_gocert_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network([], [], [])}, + def test_given_only_database_storage_container_can_connect_network_not_available_notary_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("storages not yet available") - def test_given_storages_available_container_can_connect_network_not_available_gocert_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network([], [], [])}, + def test_given_storages_available_container_can_connect_network_not_available_notary_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("certificates not yet created") - def test_given_only_config_storage_container_cant_connect_network_available_gocert_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network.default()}, + def test_given_only_config_storage_container_cant_connect_network_available_notary_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("container not yet connectable") - def test_given_only_database_storage_container_cant_connect_network_available_gocert_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network.default()}, + def test_given_only_database_storage_container_cant_connect_network_available_notary_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("container not yet connectable") - def test_given_storages_available_container_cant_connect_network_available_gocert_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network.default()}, + def test_given_storages_available_container_cant_connect_network_available_notary_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("container not yet connectable") - def test_given_only_config_storage_container_can_connect_network_available_gocert_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network.default()}, + def test_given_only_config_storage_container_can_connect_network_available_notary_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("storages not yet available") - def test_given_only_database_storage_container_can_connect_network_available_gocert_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network.default()}, + def test_given_only_database_storage_container_can_connect_network_available_notary_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("storages not yet available") - def test_given_storages_available_container_can_connect_network_available_gocert_running_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network.default()}, + def test_given_storages_available_container_can_connect_network_available_notary_running_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": False}, + **{"is_api_available.return_value": True, "is_initialized.return_value": False}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("certificates not yet created") - def test_given_only_config_storage_container_cant_connect_network_not_available_gocert_initialized_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network([], [], [])}, + def test_given_only_config_storage_container_cant_connect_network_not_available_notary_initialized_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("container not yet connectable") - def test_given_only_database_storage_container_cant_connect_network_not_available_gocert_initialized_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network([], [], [])}, + def test_given_only_database_storage_container_cant_connect_network_not_available_notary_initialized_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("container not yet connectable") - def test_given_storages_available_container_cant_connect_network_not_available_gocert_initialized_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network([], [], [])}, + def test_given_storages_available_container_cant_connect_network_not_available_notary_initialized_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("container not yet connectable") - def test_given_only_config_storage_container_can_connect_network_not_available_gocert_initialized_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network([], [], [])}, + def test_given_only_config_storage_container_can_connect_network_not_available_notary_initialized_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("storages not yet available") - def test_given_only_database_storage_container_can_connect_network_not_available_gocert_initialized_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network([], [], [])}, + def test_given_only_database_storage_container_can_connect_network_not_available_notary_initialized_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("storages not yet available") - def test_given_storages_available_container_can_connect_network_not_available_gocert_initialized_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network([], [], [])}, + def test_given_storages_available_container_can_connect_network_not_available_notary_initialized_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info", bind_addresses=[])}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("certificates not yet created") - def test_given_only_config_storage_container_cant_connect_network_available_gocert_initialized_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network.default()}, + def test_given_only_config_storage_container_cant_connect_network_available_notary_initialized_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("container not yet connectable") - def test_given_only_database_storage_container_cant_connect_network_available_gocert_initialized_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network.default()}, + def test_given_only_database_storage_container_cant_connect_network_available_notary_initialized_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("container not yet connectable") - def test_given_storages_available_container_cant_connect_network_available_gocert_initialized_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=False)], - networks={"juju-info": Network.default()}, + def test_given_storages_available_container_cant_connect_network_available_notary_initialized_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=False, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("container not yet connectable") - def test_given_only_config_storage_container_can_connect_network_available_gocert_initialized_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network.default()}, + def test_given_only_config_storage_container_can_connect_network_available_notary_initialized_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("storages not yet available") - def test_given_only_database_storage_container_can_connect_network_available_gocert_initialized_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network.default()}, + def test_given_only_database_storage_container_can_connect_network_available_notary_initialized_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("storages not yet available") - def test_given_storages_available_container_can_connect_network_available_gocert_initialized_when_collect_status_then_status_is_waiting( - self, context - ): - state = State( - storage=[Storage(name="config"), Storage(name="database")], - containers=[Container(name="gocert", can_connect=True)], - networks={"juju-info": Network.default()}, + def test_given_storages_available_container_can_connect_network_available_notary_initialized_when_collect_status_then_status_is_waiting( + self, context + ): + state = State( + storages={Storage(name="config"), Storage(name="database")}, + containers={ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + }, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert", + "notary.Notary", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.WaitingStatus("certificates not yet created") - def test_given_gocert_available_and_initialized_when_collect_status_then_status_is_active( + def test_given_notary_available_and_initialized_when_collect_status_then_status_is_active( self, context ): with tempfile.TemporaryDirectory() as tempdir: - config_mount = Mount("/etc/gocert/config", tempdir) + config_mount = Mount(location="/etc/notary/config", source=tempdir) state = State( - storage=[Storage(name="config"), Storage(name="database")], + storages={Storage(name="config"), Storage(name="database")}, containers=[ - Container(name="gocert", can_connect=True, mounts={"config": config_mount}) + Container(name="notary", can_connect=True, mounts={"config": config_mount}) ], - networks={"juju-info": Network.default()}, + networks={Network("juju-info")}, leader=True, ) @@ -1436,30 +2953,30 @@ def test_given_gocert_available_and_initialized_when_collect_status_then_status_ f.write(str(certificate)) with patch( - "gocert.GoCert.__new__", + "notary.Notary.__new__", return_value=Mock( - **{"is_api_available.return_value": True, "is_initialized.return_value": True}, + **{"is_api_available.return_value": True, "is_initialized.return_value": True}, # type: ignore ), ): - out = context.run(Event("collect-unit-status"), state) + out = context.run(context.on.collect_unit_status(), state) assert out.unit_status == ops.ActiveStatus() - def test_given_gocert_available_and_not_initialized_when_configure_then_admin_user_created( + def test_given_notary_available_and_not_initialized_when_configure_then_admin_user_created( self, context ): with tempfile.TemporaryDirectory() as tempdir: - config_mount = Mount("/etc/gocert/config", tempdir) + config_mount = Mount(location="/etc/notary/config", source=tempdir) state = State( - storage=[Storage(name="config"), Storage(name="database")], + storages={Storage(name="config"), Storage(name="database")}, containers=[ - Container(name="gocert", can_connect=True, mounts={"config": config_mount}) + Container(name="notary", can_connect=True, mounts={"config": config_mount}) ], - networks={"juju-info": Network.default()}, + networks={Network("juju-info")}, leader=True, ) with patch( - "gocert.GoCert.__new__", + "notary.Notary.__new__", return_value=Mock( **{ "is_api_available.return_value": True, @@ -1469,7 +2986,419 @@ def test_given_gocert_available_and_not_initialized_when_configure_then_admin_us }, ), ): - out = context.run(Event("update-status"), state) + out = context.run(context.on.update_status(), state) assert len(out.secrets) == 1 - assert out.secrets[0].label == "GoCert Login Details" - assert out.secrets[0].contents[1].get("token") == "example-token" + secret = out.get_secret(label="Notary Login Details") + assert secret.latest_content.get("token") == "example-token" + + def test_given_tls_requirer_available_when_notary_unreachable_then_no_error_raised( + self, context + ): + state = State( + storages={Storage(name="config"), Storage(name="database")}, + containers=[ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + ], + networks={Network("juju-info")}, + leader=True, + relations=[Relation(id=1, endpoint=CERTIFICATE_PROVIDER_RELATION_NAME)], + ) + with patch( + "notary.Notary.__new__", + return_value=Mock( + **{ + "is_api_available.return_value": True, + "is_initialized.return_value": False, + "login.return_value": "example-token", + "token_is_valid.return_value": False, + }, + ), + ): + context.run(context.on.update_status(), state) + + @patch(f"{TLS_LIB_PATH}.TLSCertificatesProvidesV4.get_certificate_requests") + def test_given_tls_requirer_available_when_configure_then_csrs_posted_to_notary( + self, mock_get_certificate_requests, context + ): + state = State( + storages={Storage(name="config"), Storage(name="database")}, + containers=[ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + ], + networks={Network("juju-info")}, + leader=True, + relations=[Relation(id=1, endpoint=CERTIFICATE_PROVIDER_RELATION_NAME)], + secrets={ + Secret( + {"username": "hello", "password": "world", "token": "test-token"}, + id="1", + label=NOTARY_LOGIN_SECRET_LABEL, + owner="app", + ) + }, + ) + csr = generate_csr(private_key=generate_private_key(), common_name="me") + mock_get_certificate_requests.return_value = [ + RequirerCSR( + relation_id=1, + certificate_signing_request=csr, + ) + ] + post_call = Mock() + with patch( + "notary.Notary.__new__", + return_value=Mock( + **{ + "is_api_available.return_value": True, + "is_initialized.return_value": True, + "token_is_valid.return_value": True, + "get_certificate_requests_table.return_value": CertificateRequests(rows=[]), + "post_csr": post_call, + }, + ), + ): + context.run(context.on.update_status(), state) + + post_call.assert_called_once_with(str(csr), "test-token") + + @patch(f"{TLS_LIB_PATH}.TLSCertificatesProvidesV4.get_certificate_requests") + def test_given_tls_requirers_available_when_csrs_already_posted_then_duplicate_csr_not_posted( + self, mock_get_certificate_requests, context + ): + state = State( + storages={Storage(name="config"), Storage(name="database")}, + containers=[ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + ], + networks={Network("juju-info")}, + leader=True, + relations=[Relation(id=1, endpoint=CERTIFICATE_PROVIDER_RELATION_NAME)], + secrets={ + Secret( + {"username": "hello", "password": "world", "token": "test-token"}, + id="1", + label=NOTARY_LOGIN_SECRET_LABEL, + owner="app", + ) + }, + ) + csr = generate_csr(private_key=generate_private_key(), common_name="me") + mock_get_certificate_requests.return_value = [ + RequirerCSR( + relation_id=1, + certificate_signing_request=csr, + ) + ] + post_call = Mock() + with patch( + "notary.Notary.__new__", + return_value=Mock( + **{ + "is_api_available.return_value": True, + "is_initialized.return_value": True, + "token_is_valid.return_value": True, + "get_certificate_requests_table.return_value": CertificateRequests( + rows=[CertificateRequest(id=1, csr=str(csr), certificate_chain="")] + ), + "post_csr": post_call, + }, + ), + ): + context.run(context.on.update_status(), state) + + post_call.assert_not_called() + + @patch(f"{TLS_LIB_PATH}.TLSCertificatesProvidesV4.set_relation_certificate") + @patch(f"{TLS_LIB_PATH}.TLSCertificatesProvidesV4.get_certificate_requests") + def test_given_tls_requirers_available_when_certificate_available_then_certs_provided_to_requirer( + self, mock_get_certificate_requests, mock_set_relation_certificate, context + ): + state = State( + storages={Storage(name="config"), Storage(name="database")}, + containers=[ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + ], + networks={Network("juju-info")}, + leader=True, + relations=[Relation(id=1, endpoint=CERTIFICATE_PROVIDER_RELATION_NAME)], + secrets={ + Secret( + {"username": "hello", "password": "world", "token": "test-token"}, + id="1", + label=NOTARY_LOGIN_SECRET_LABEL, + owner="app", + ) + }, + ) + ca_pk = generate_private_key() + ca = generate_ca(ca_pk, 365, "me") + csr = generate_csr(private_key=generate_private_key(), common_name="notary.com") + cert = generate_certificate(csr, ca, ca_pk, 365) + mock_get_certificate_requests.return_value = [ + RequirerCSR( + relation_id=1, + certificate_signing_request=csr, + ) + ] + with patch( + "notary.Notary.__new__", + return_value=Mock( + **{ + "is_api_available.return_value": True, + "is_initialized.return_value": True, + "token_is_valid.return_value": True, + "get_certificate_requests_table.return_value": CertificateRequests( + rows=[ + CertificateRequest( + id=1, csr=str(csr), certificate_chain=[str(cert), str(ca)] + ) + ] + ), + }, + ), + ): + context.run(context.on.update_status(), state) + mock_set_relation_certificate.assert_called_once() + + @patch(f"{TLS_LIB_PATH}.TLSCertificatesProvidesV4.get_issued_certificates") + @patch(f"{TLS_LIB_PATH}.TLSCertificatesProvidesV4.set_relation_certificate") + @patch(f"{TLS_LIB_PATH}.TLSCertificatesProvidesV4.get_certificate_requests") + def test_given_tls_requirers_when_invalid_certificate_available_when_configure_then_new_cert_provided( + self, + mock_get_certificate_requests, + mock_set_relation_certificate, + mock_get_issued_certificates, + context, + ): + state = State( + storages={Storage(name="config"), Storage(name="database")}, + containers=[ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + ], + networks={Network("juju-info")}, + leader=True, + relations=[Relation(id=1, endpoint=CERTIFICATE_PROVIDER_RELATION_NAME)], + secrets={ + Secret( + {"username": "hello", "password": "world", "token": "test-token"}, + id="1", + label=NOTARY_LOGIN_SECRET_LABEL, + owner="app", + ) + }, + ) + ca_pk = generate_private_key() + ca = generate_ca(ca_pk, 365, "me") + csr = generate_csr(private_key=generate_private_key(), common_name="notary.com") + old_cert = generate_certificate(csr, ca, ca_pk, 365) + new_cert = generate_certificate(csr, ca, ca_pk, 366) + mock_get_certificate_requests.return_value = [ + RequirerCSR( + relation_id=1, + certificate_signing_request=csr, + ) + ] + mock_get_issued_certificates.return_value = [ + ProviderCertificate( + relation_id=1, + certificate_signing_request=csr, + certificate=old_cert, + ca=ca, + chain=[old_cert, ca], + ) + ] + with patch( + "notary.Notary.__new__", + return_value=Mock( + **{ + "is_api_available.return_value": True, + "is_initialized.return_value": True, + "token_is_valid.return_value": True, + "get_certificate_requests_table.return_value": CertificateRequests( + rows=[ + CertificateRequest( + id=1, csr=str(csr), certificate_chain=[str(new_cert), str(ca)] + ) + ] + ), + }, + ), + ): + context.run(context.on.update_status(), state) + mock_set_relation_certificate.assert_called_once() + + @patch(f"{TLS_LIB_PATH}.TLSCertificatesProvidesV4.get_issued_certificates") + @patch(f"{TLS_LIB_PATH}.TLSCertificatesProvidesV4.set_relation_certificate") + @patch(f"{TLS_LIB_PATH}.TLSCertificatesProvidesV4.get_certificate_requests") + def test_given_certificate_rejected_in_notary_when_configure_then_certificate_revoked( + self, + mock_get_certificate_requests, + mock_set_relation_certificate, + mock_get_issued_certificates, + context, + ): + state = State( + storages={Storage(name="config"), Storage(name="database")}, + containers=[ + Container( + name="notary", + can_connect=True, + layers={ + "notary": Layer( + { + "summary": "notary layer", + "description": "pebble config layer for notary", + "services": { + "notary": { + "override": "replace", + "summary": "notary", + "command": "notary -config /etc/notary/config/config.yaml", + "startup": "enabled", + } + }, + } + ) + }, + ) + ], + networks={Network("juju-info")}, + leader=True, + relations=[Relation(id=1, endpoint=CERTIFICATE_PROVIDER_RELATION_NAME)], + secrets=[ + Secret( + {"username": "hello", "password": "world", "token": "test-token"}, + id="1", + label=NOTARY_LOGIN_SECRET_LABEL, + owner="app", + ) + ], + ) + ca_pk = generate_private_key() + ca = generate_ca(ca_pk, 365, "me") + csr = generate_csr(private_key=generate_private_key(), common_name="notary.com") + old_cert = generate_certificate(csr, ca, ca_pk, 365) + mock_get_certificate_requests.return_value = [ + RequirerCSR( + relation_id=1, + certificate_signing_request=csr, + ) + ] + mock_get_issued_certificates.return_value = [ + ProviderCertificate( + relation_id=1, + certificate_signing_request=csr, + certificate=old_cert, + ca=ca, + chain=[old_cert, ca], + ) + ] + with patch( + "notary.Notary.__new__", + return_value=Mock( + **{ + "is_api_available.return_value": True, + "is_initialized.return_value": True, + "token_is_valid.return_value": True, + "get_certificate_requests_table.return_value": CertificateRequests( + rows=[CertificateRequest(id=1, csr=str(csr), certificate_chain="rejected")] + ), + }, + ), + ): + context.run(context.on.update_status(), state) + mock_set_relation_certificate.assert_called_once()