generated from coulomb/repo-seed
tegwick
39080333da
Extract bootstrap logic into src/ops_hub/bootstrap.py with dry-run support, public hub discovery, and unit tests. Update the gate probe to accept public hub listing (200 or 401), document dev commands and architecture, and archive finished OPS-WP-0001 and OPS-WP-0002 workplans.
96 lines
3.5 KiB
Python
96 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from typing import Any
|
|
|
|
from ops_hub.bootstrap import (
|
|
MANIFEST_PATH,
|
|
WIDGETS_PATH,
|
|
find_hub,
|
|
load_manifest,
|
|
load_widgets,
|
|
manifest_body,
|
|
plan_api_consumer,
|
|
plan_event,
|
|
plan_hub,
|
|
plan_runtime_key,
|
|
)
|
|
|
|
|
|
class BootstrapTests(unittest.TestCase):
|
|
def test_load_manifest_and_widgets(self) -> None:
|
|
manifest = load_manifest()
|
|
widgets = load_widgets()
|
|
self.assertEqual(manifest["hub"]["slug"], "ops-hub")
|
|
self.assertGreaterEqual(len(widgets), 1)
|
|
|
|
def test_manifest_body_includes_hub_id_when_provided(self) -> None:
|
|
manifest = load_manifest()
|
|
body = manifest_body(manifest, "hub-123")
|
|
self.assertEqual(body["hubId"], "hub-123")
|
|
self.assertIn("declaredWidgetTypes", body)
|
|
|
|
def test_find_hub_uses_public_discovery_before_operator_token(self) -> None:
|
|
calls: list[str | None] = []
|
|
|
|
def fake_list_items(base_url: str, path: str, token: str | None) -> list[dict[str, Any]]:
|
|
calls.append(token)
|
|
if token is None:
|
|
return [{"slug": "ops-hub", "id": "hub-public"}]
|
|
return []
|
|
|
|
with patch("ops_hub.bootstrap.list_items", side_effect=fake_list_items):
|
|
found = find_hub("https://hub.example", "operator-key", "ops-hub")
|
|
self.assertEqual(found["id"], "hub-public")
|
|
self.assertEqual(calls, [None])
|
|
|
|
def test_plan_hub_reports_create_when_missing(self) -> None:
|
|
manifest = load_manifest()
|
|
with patch("ops_hub.bootstrap.find_hub", return_value=None):
|
|
plan = plan_hub("https://hub.example", "operator-key", manifest)
|
|
self.assertEqual(plan["action"], "create")
|
|
|
|
def test_plan_hub_reports_reuse_when_present(self) -> None:
|
|
manifest = load_manifest()
|
|
existing = {"id": "hub-1", "slug": "ops-hub"}
|
|
with patch("ops_hub.bootstrap.find_hub", return_value=existing):
|
|
plan = plan_hub("https://hub.example", "operator-key", manifest)
|
|
self.assertEqual(plan["action"], "reuse")
|
|
self.assertEqual(plan["record"]["id"], "hub-1")
|
|
|
|
def test_plan_api_consumer_reports_reuse(self) -> None:
|
|
with patch(
|
|
"ops_hub.bootstrap.list_items",
|
|
return_value=[{"id": "consumer-1", "name": "ops-hub"}],
|
|
):
|
|
plan = plan_api_consumer("https://hub.example", "operator-key")
|
|
self.assertEqual(plan["action"], "reuse")
|
|
|
|
def test_plan_runtime_key_reports_reuse_from_env(self) -> None:
|
|
with patch.dict("os.environ", {"OPS_HUB_KEY": "runtime-key-value"}, clear=False):
|
|
plan = plan_runtime_key("consumer-1")
|
|
self.assertEqual(plan["action"], "reuse")
|
|
self.assertEqual(plan["keyPrefix"], "runtime-")
|
|
|
|
def test_plan_event_targets_gitea_readiness_widget(self) -> None:
|
|
widgets = {
|
|
"reuse": [{"id": "widget-1", "capabilityRef": "ops:readiness:gitea-registry"}],
|
|
"create": [],
|
|
}
|
|
event = plan_event(widgets, skip_event=False)
|
|
self.assertIsNotNone(event)
|
|
assert event is not None
|
|
self.assertEqual(event["widgetId"], "widget-1")
|
|
self.assertEqual(event["eventType"], "ops-endpoint-verified")
|
|
|
|
def test_seed_files_are_valid_json(self) -> None:
|
|
json.loads(MANIFEST_PATH.read_text(encoding="utf-8"))
|
|
json.loads(WIDGETS_PATH.read_text(encoding="utf-8"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |