test: add COS Lite deployment and Loki integration tests for k8s-5-observe

examples/k8s-5-observe/pyproject.toml

@@ -43,6 +43,7 @@ integration = [
     "jubilant",
     "pytest",
     "PyYAML",
+    "requests",
 ]
 
 # Testing tools configuration

examples/k8s-5-observe/tests/integration/conftest.py

@@ -40,6 +40,20 @@ def juju(request: pytest.FixtureRequest):
             print(log, end="", file=sys.stderr)
 
 
+@pytest.fixture(scope="module")
+def cos_juju(request: pytest.FixtureRequest):
+    """Create a temporary Juju model for COS Lite."""
+    with jubilant.temp_model() as juju:
+        juju.wait_timeout = 10 * 60
+        yield juju
+
+        if request.session.testsfailed:
+            logger.info("Collecting Juju logs from COS Lite model...")
+            time.sleep(0.5)  # Wait for Juju to process logs.
+            log = juju.debug_log(limit=1000)
+            print(log, end="", file=sys.stderr)
+
+
 @pytest.fixture(scope="session")
 def charm():
     """Return the path of the charm under test."""

examples/k8s-5-observe/tests/integration/test_charm.py

@@ -15,10 +15,12 @@
 # The integration tests use the Jubilant library. See https://documentation.ubuntu.com/jubilant/
 # To learn more about testing, see https://documentation.ubuntu.com/ops/latest/explanation/testing/
 
+import json
 import logging
 import pathlib
 
 import jubilant
+import requests
 import yaml
 
 logger = logging.getLogger(__name__)
@@ -49,3 +51,39 @@ def test_database_integration(juju: jubilant.Juju):
     juju.deploy("postgresql-k8s", channel="14/stable", trust=True)
     juju.integrate(APP_NAME, "postgresql-k8s")
     juju.wait(jubilant.all_active)
+
+
+def test_cos_lite(cos_juju: jubilant.Juju):
+    """Deploy COS Lite and verify a Loki offer is created.
+
+    Waits for all COS Lite applications to settle down, then offers the
+    Loki logging endpoint for cross-model integration.
+    """
+    cos_juju.deploy("cos-lite", trust=True)
+    cos_juju.wait(jubilant.all_active)
+
+    cos_juju.offer("loki", endpoint="logging")
+
+    # Verify the Loki offer was created; raises CLIError if not found.
+    cos_juju.cli("find-offers", "--interface", "loki_push_api", include_model=False)
+
+
+def test_loki_integration(juju: jubilant.Juju, cos_juju: jubilant.Juju):
+    """Verify that the charm integrates with Loki.
+
+    Assert that the charm remains active after the Loki integration is established,
+    and that Loki has registered the application as a log source.
+    """
+    juju.integrate(APP_NAME, f"admin/{cos_juju.model}.loki")
+    juju.wait(jubilant.all_active)
+
+    # Find the Loki API base URL via Traefik's show-proxied-endpoints action.
+    task = cos_juju.run("traefik/0", "show-proxied-endpoints")
+    endpoints = json.loads(task.results["proxied-endpoints"])
+    loki_url = endpoints["loki/0"]["url"]
+
+    # Confirm that our application appears in Loki's label values.
+    api_url = f"{loki_url}/loki/api/v1/label/juju_application/values"
+    response = requests.get(api_url)
+    response.raise_for_status()
+    assert APP_NAME in response.json()["data"]