diff --git a/my_egeria/my_egeria/DemoCode/My_Profile/CreateSubscriptionRequestScreen.py b/my_egeria/my_egeria/DemoCode/My_Profile/CreateSubscriptionRequestScreen.py index 795f6eb7..50ac7045 100644 --- a/my_egeria/my_egeria/DemoCode/My_Profile/CreateSubscriptionRequestScreen.py +++ b/my_egeria/my_egeria/DemoCode/My_Profile/CreateSubscriptionRequestScreen.py @@ -90,6 +90,7 @@ def action_create_subscription(self): if self.selected_item: self.response["externalSourceGUID"] = self.selected_item self.response["guid"] = self.selected_item + self.response["GUID"] = self.selected_item if self.display_name: self.response["displayName"] = self.display_name if self.status: diff --git a/my_egeria/my_egeria/DemoCode/My_Profile/MainScreen.py b/my_egeria/my_egeria/DemoCode/My_Profile/MainScreen.py index 8fcb2f31..bfb80836 100644 --- a/my_egeria/my_egeria/DemoCode/My_Profile/MainScreen.py +++ b/my_egeria/my_egeria/DemoCode/My_Profile/MainScreen.py @@ -52,12 +52,12 @@ def compose(self) -> ComposeResult: Static(f"Other Functions"), Static(f"[b]Select a function[/b]"), OptionList( - Option("User Identities"), - Option("Catalogs/Shop for Data"), - Option("Edit Profile"), - Option("Subscriptions"), - Option("Technology Types"), - Option("User Bookmarks"), + Option("[] User Identities"), + Option("[] Catalogs/Shop for Data"), + Option("[] Edit Profile"), + Option("[] Subscriptions"), + Option("[] Technology Types"), + Option("[] User Bookmarks"), id="other_function_list" ), id="other_function_container" diff --git a/my_egeria/my_egeria/DemoCode/My_Profile/my_profile_app.py b/my_egeria/my_egeria/DemoCode/My_Profile/my_profile_app.py index 9a8d98ea..bef8f6ad 100644 --- a/my_egeria/my_egeria/DemoCode/My_Profile/my_profile_app.py +++ b/my_egeria/my_egeria/DemoCode/My_Profile/my_profile_app.py @@ -459,7 +459,7 @@ async def _populate_tables(self) -> Any: str(p.get("Description", "")), str(p.get("GUID", p.get("guid", ""))), ) - self.projects_table.loading=False + self.projects_table.loading=False if self.communities_table: for c in self.communities if isinstance(self.communities, list) else []: self.communities_table.add_row( @@ -468,7 +468,7 @@ async def _populate_tables(self) -> Any: str(c.get("Description", "")), str(c.get("GUID", c.get("guid", ""))), ) - self.communities_table.loading=False + self.communities_table.loading=False for r in self.roles if isinstance(self.roles, list) else []: self.roles_table.add_row( str(r.get("Name", "")), @@ -476,7 +476,7 @@ async def _populate_tables(self) -> Any: str(r.get("Description", "")), str(r.get("GUID", r.get("guid", ""))), ) - self.roles_table.loading=False + self.roles_table.loading=False for t in self.teams if isinstance(self.teams, list) else []: self.teams_table.add_row( str(t.get("Assignment Type", "")), @@ -484,7 +484,7 @@ async def _populate_tables(self) -> Any: str(t.get("Description", "")), str(t.get("GUID", t.get("guid", ""))), ) - self.teams_table.loading=False + self.teams_table.loading=False for b in self.blogs if isinstance(self.blogs, list) else []: self.blogs_table.add_row( str(b.get("qualifiedName", "")), @@ -492,7 +492,7 @@ async def _populate_tables(self) -> Any: str(b.get("text", "")), str(b.get("GUID", "")), ) - self.blogs_table.loading=False + self.blogs_table.loading=False for j in self.journal if isinstance(self.journal, list) else []: self.journal_table.add_row( str(j.get("qualifiedName", "")), @@ -500,7 +500,7 @@ async def _populate_tables(self) -> Any: str(j.get("text", "")), str(j.get("GUID", j.get("guid", ""))), ) - self.journal_table.loading=False + self.journal_table.loading=False for td in self.todos if isinstance(self.todos, list) else []: self.todos_table.add_row( str(td.get("Name", "")), @@ -508,7 +508,7 @@ async def _populate_tables(self) -> Any: str(td.get("Description", "")), str(td.get("GUID", td.get("guid", ""))), ) - self.todos_table.loading=False + self.todos_table.loading=False for ui in self.user_identity if isinstance(self.user_identity, list) else []: self.user_identity_table.add_row( str(ui.get("Display Name", "")), @@ -516,7 +516,7 @@ async def _populate_tables(self) -> Any: str(ui.get("Distinguished Name", "")), str(ui.get("GUID", ui.get("guid", ""))), ) - self.user_identity_table.loading=False + self.user_identity_table.loading=False for c in self.communities if isinstance(self.communities, list) else []: self.associations_table.add_row( str(c.get("Assignment Type", "")), @@ -524,7 +524,7 @@ async def _populate_tables(self) -> Any: str(c.get("Description", "")), str(c.get("GUID", c.get("guid", ""))), ) - self.associations_table.loading=False + self.associations_table.loading=False def action_quit(self) -> Any: self.exit(200) @@ -537,7 +537,7 @@ async def action_refresh(self) -> None: @on(OptionList.OptionSelected, "#other_function_list") async def handle_option_selected(self, event: OptionList.OptionSelected) -> None: - selected_option = event.option.prompt + selected_option = event.option.prompt.strip("[] ") selected_option_id = event.option.id self.log(f"Selected option: {selected_option} ({selected_option_id})") if selected_option == "Technology Types": diff --git a/my_egeria/my_egeria/DemoCode/My_Profile/shop_for_data_handler.py b/my_egeria/my_egeria/DemoCode/My_Profile/shop_for_data_handler.py index c3ff91f4..9d515fac 100644 --- a/my_egeria/my_egeria/DemoCode/My_Profile/shop_for_data_handler.py +++ b/my_egeria/my_egeria/DemoCode/My_Profile/shop_for_data_handler.py @@ -911,12 +911,13 @@ def create_subscription_callback(self, result: Any) -> None: item_guid = ( result.get("externalSourceGUID") or result.get("guid") + or result.get("GUID") or (self.selected_item if hasattr(self, "selected_item") else "") or "" ) if isinstance(result, dict) else (self.selected_item if hasattr(self, "selected_item") else "") body = { - "class": "NewAgreementRequestBody", + "class": "NewElementRequestBody", "isOwnAnchor": True, "anchorScopeGUID": None, "parentGUID": None, @@ -927,18 +928,10 @@ def create_subscription_callback(self, result: Any) -> None: "qualifiedName": "DigitalSubscription::" + display_name, "displayName": display_name or "display name", "description": description, - "userDefinedStatus": "DRAFT", + "contentStatus": "ACTIVE", "identifier": identifier, - "supportLevel": "Community", - "serviceLevels": None, - "additionalProperties": None, - }, - "initialStatus": status, - "externalSourceGUID": item_guid, - "externalSourceName": display_name, - "effectiveTime": None, - "forLineage": False, - "forDuplicateProcessing": False, + "supportLevel": "Community" + } } try: diff --git a/my_egeria/my_egeria/DemoCode/My_Profile/tech_types_handler.py b/my_egeria/my_egeria/DemoCode/My_Profile/tech_types_handler.py index d7b8290c..b1dec5d1 100644 --- a/my_egeria/my_egeria/DemoCode/My_Profile/tech_types_handler.py +++ b/my_egeria/my_egeria/DemoCode/My_Profile/tech_types_handler.py @@ -305,13 +305,15 @@ def tech_type_templates_callback(self, result: Any) -> Any: "class": "TemplateRequestBody", "externalSourceGUID": self.full_template.get("externalSourceGUID") or "", "externalSourceName": self.full_template.get("externalSourceName") or "", - "typeName": self.full_template.get("typeName") or "", - "templateGUID": self.full_template.get("Catalog Template GUID"), + "templateGUID": self.full_template.get("templateGUID"), "anchorGUID": self.full_template.get("anchorGUID"), "isOwnAnchor": "false", "effectiveFrom": "2026-01-01", "effectiveTo": "2030-12-31", - "replacementProperties": self.full_template.get("replacementProperties") or {}, + # Must be absent rather than {}: the server resolves this to a typed + # EntityProperties subtype and rejects an untyped empty object with + # "missing type id property 'class'". + "replacementProperties": self.full_template.get("replacementProperties") or None, "placeholderPropertyValues": {}, "parentGUID": None, "parentRelationshipTypeName": None, @@ -327,7 +329,7 @@ def tech_type_templates_callback(self, result: Any) -> Any: try: tokendata = self.autoc.create_egeria_bearer_token(self.user_name, self.user_password) my_md_instance = AutomatedCuration(self.view_server, self.platform_url, self.user_name, self.user_password, tokendata) - new_guid = my_md_instance.initiate_gov_action_process(body=request_body) + new_guid = my_md_instance.create_elem_from_template(body=request_body) except Exception as e: self.log(f"Exception in create_element_from_template: {e}") if isinstance(e, PyegeriaException): diff --git a/tests/micro-tests/my_profile/README.md b/tests/micro-tests/my_profile/README.md new file mode 100644 index 00000000..0f4c4ae8 --- /dev/null +++ b/tests/micro-tests/my_profile/README.md @@ -0,0 +1,149 @@ + + + +# tests/micro-tests/my_profile + +Tests for the `MyProfileApp` Textual TUI in +`my_egeria/my_egeria/DemoCode/My_Profile/` — not the `pyegeria.omvs.my_profile` +OMVS client, which is covered in `tests/functional-tests/`. + +The suite runs two ways from the same test code. + +```bash +# Fake backend (default) — no server needed, nothing is written anywhere +pytest tests/micro-tests/my_profile/ + +# Live backend — real Egeria at https://localhost:9443 +PYEG_LIVE_EGERIA=1 pytest tests/micro-tests/my_profile/ + +# Just the tests that actually switch +PYEG_LIVE_EGERIA=1 pytest tests/micro-tests/my_profile/ -m live_capable +``` + +## ⚠️ Live mode writes to your Egeria instance + +Live mode is not read-only. It really does create digital subscriptions, +really does initiate a governance action process, and really would create a +personal profile if the configured user did not already have one. Point it at +a development instance, not anything you care about. + +## Configuration + +| Variable | Default | Purpose | +|---|---|---| +| `PYEG_LIVE_EGERIA` | unset (fake) | `1`/`true`/`yes`/`on` switches the suite to the live backend | +| `PYEG_PLATFORM_URL` | `https://localhost:9443` | Egeria platform URL | +| `PYEG_SERVER_NAME` | `qs-view-server` | View server name | +| `PYEG_USER_ID` | `garygeeke` | User to authenticate as | +| `PYEG_USER_PWD` | `secret` | Password | + +In live mode these are also exported as `EGERIA_PLATFORM_URL` / +`EGERIA_VIEW_SERVER` / `EGERIA_USER` / `EGERIA_USER_PASSWORD` during +`pytest_configure`, so the app's own `load_app_config()` resolves to the same +server. That has to happen before collection finishes, because `load_app_config()` +caches on first call. + +If `PYEG_LIVE_EGERIA=1` is set but the server will not issue a bearer token, +every live-capable test skips with the reason. The server is probed once per +session, so an unreachable server costs a fraction of a second rather than a +screenful of connection errors. + +## Fake mode never touches the network + +An autouse fixture blocks `httpx` for the whole folder whenever +`PYEG_LIVE_EGERIA` is off, so a fake-mode test that leaves an Egeria call +unmocked fails immediately: + +``` +UnexpectedNetworkCall: test_foo tried to reach https://localhost:9443/... while +running against fakes. Patch the Egeria client/function through the `backend` +fixture (backend.patch / backend.always_fake), or mark the test +`@pytest.mark.allow_network` if the call is genuinely intended. +``` + +This is what keeps the suite consistent. Before the guard existed, four tests +quietly reached a real server in "fake" mode — usually because the behaviour +under test was mocked but something *adjacent* was not (entering +`MyProfileApp.run_test()` runs `on_mount`, which builds a real `MyProfile` +client regardless of what the test itself patched). Those tests passed on a +machine with a server running and behaved differently on one without. + +`stub_profile_client()` in `test_my_profile_app.py` is the helper for that +specific case: any test that enters `app.run_test()` should call it. + +## How the switch works + +`egeria_backend.py` provides the `backend` fixture. The important method is +`backend.patch(target, returns=...)`: + +- **fake mode** — replaces `target` with `MagicMock(return_value=returns)` +- **live mode** — replaces it with `MagicMock(wraps=)`, so the + call reaches Egeria *and* the mock still records it + +Because a wrapping mock records calls, assertions about **how** the app called +Egeria (report-spec names, params, verbs) are written once and hold in both +modes. Only assertions about the returned **data** need to differ, and those go +through `backend.expect(actual, fake=..., live=)`: exact equality +against the fixture in fake mode, a structural check in live mode. + +Other helpers: + +| Helper | Use | +|---|---| +| `backend.always_fake(target, ...)` | Paths a live server cannot produce on demand — injected exceptions, deliberately empty result sets. These test the app's own error handling, so they stay faked in both modes. | +| `backend.unique(name)` | Suffixes a name with a random hex in live mode only. Egeria enforces `qualifiedName` uniqueness, so a live write test reusing a fixed name passes once and then 409s forever. | +| `backend.apply_connection(app)` | Points a `Dummy*App` test harness at the active backend. | +| `backend.patch_object(cls, attr, repl)` | Non-Egeria patching (UI widgets) — same in both modes. | + +Live-only fixtures discover real inputs from the server rather than using +synthetic GUIDs, and skip with a specific reason when the server has no +suitable data: `live_my_profile`, `live_team_role_name`, `live_tech_type_name`, +`live_catalog_template`. + +## Adding a test that touches Egeria + +1. Take the `backend` fixture and mark the test `@pytest.mark.live_capable`. +2. Replace `@patch("module.symbol")` with + `backend.patch("module.symbol", returns=)`. +3. Keep call-args assertions unconditional; route value assertions through + `backend.expect(...)`. +4. If the test needs a specific element to exist, add a live-discovery fixture + rather than hardcoding a GUID. +5. If the path is an injected error or an empty result, use + `backend.always_fake(...)` and leave the `live_capable` marker off. +6. If the test enters `MyProfileApp.run_test()`, call `stub_profile_client(...)` + so `on_mount`'s profile load is mocked too. + +The network guard will tell you if you miss one. + +## Known live-mode outcomes + +- `test_create_profile_screen_create_profile` **skips** live when the configured + user already has a profile. `add_my_profile` creates the profile for the + *calling* user and the server rejects a second one + (`OMVS-MY-PROFILE-400-001`), so the path is only reachable for a user who has + none yet. +- `test_tech_type_templates_callback_success` **creates a real metadata element** + from a real catalog template on every live run, and deliberately does not + clean it up — the target is a disposable test instance that gets reloaded from + scratch regularly, so teardown would add failure modes for no benefit. If you + ever do need to prune, search display names for the `pytest-` prefix. + +## What live mode has caught so far + +Worth knowing, because it is the argument for running live periodically. +`tech_type_templates_callback` had three stacked defects that full mocking hid +completely — the mocked `AutomatedCuration` accepted a nonexistent method called +with a malformed body built from keys that do not exist: + +1. it called `initiate_gov_action_process(body=...)`, which takes + `action_type_qualified_name` and has no `body` parameter → `TypeError`; +2. it read `full_template.get("Catalog Template GUID")`, a key that appears + neither in the server payload nor anywhere in the codebase → `templateGUID` + was always `None`; +3. it sent `replacementProperties: {}`, which the server rejects with + *"missing type id property 'class'"* — the field must be absent. + +All three are fixed in `tech_types_handler.py`. The lesson for new tests: an +assertion against a `MagicMock` proves the app called *something*, not that it +called the right thing with a body the server accepts. diff --git a/tests/micro-tests/my_profile/conftest.py b/tests/micro-tests/my_profile/conftest.py index e8093659..77cc2d20 100644 --- a/tests/micro-tests/my_profile/conftest.py +++ b/tests/micro-tests/my_profile/conftest.py @@ -3,22 +3,224 @@ Copyright Contributors to the ODPi Egeria project. Fixtures and mock data for My Profile App test suite. + + The suite runs against in-memory fakes by default. Set ``PYEG_LIVE_EGERIA=1`` + to run the same tests against a real Egeria view server — see + ``egeria_backend.py`` and this folder's README for the details. """ +import contextlib import sys +import uuid from pathlib import Path -from unittest.mock import MagicMock import pytest -# Ensure repo root and My_Profile folder are in sys.path +# Ensure repo root, this folder, and the My_Profile folder are in sys.path root_path = Path(__file__).resolve().parents[3] if str(root_path) not in sys.path: sys.path.insert(0, str(root_path)) +# This folder holds egeria_backend.py, imported by conftest and the test modules. +here = Path(__file__).resolve().parent +if str(here) not in sys.path: + sys.path.insert(0, str(here)) + profile_path = root_path / "my_egeria" / "my_egeria" / "DemoCode" / "My_Profile" if str(profile_path) not in sys.path: sys.path.insert(0, str(profile_path)) +# Local to this folder; imported after sys.path is prepared above. +from egeria_backend import ( # noqa: E402 + EgeriaBackend, + block_network, + connection_settings, + export_pyegeria_env, + live_requested, + live_unavailable_reason, +) + + +def pytest_configure(config): + """Register the suite's marker and, in live mode, retarget pyegeria's config. + + ``load_app_config()`` caches the first time an app or screen is built, so the + environment has to be set here — before collection finishes — rather than in + a fixture. + """ + config.addinivalue_line( + "markers", + "live_capable: exercises Egeria and switches between fake and live backends", + ) + config.addinivalue_line( + "markers", + "allow_network: permit real HTTP even in fake mode (escape hatch)", + ) + if live_requested(): + export_pyegeria_env() + + +@pytest.fixture(autouse=True) +def no_accidental_network(request, monkeypatch): + """In fake mode, no test may reach a real server. + + Applied to every test in this folder so the suite is consistent: a test + either mocks its Egeria calls or runs live deliberately. Opt out with + `@pytest.mark.allow_network`. + """ + if live_requested() or request.node.get_closest_marker("allow_network"): + return + block_network(monkeypatch, request.node.name) + + +@pytest.fixture(scope="session") +def egeria_connection(): + """The Egeria connection details the suite is configured to use.""" + return connection_settings() + + +@pytest.fixture +def backend(request): + """Fake or live Egeria backend for a single test. + + In live mode the server is probed once per session; if it is not reachable + or will not issue a token, tests skip with the reason rather than failing + with a wall of connection errors. + """ + live = live_requested() + settings = connection_settings() + + if live: + reason = live_unavailable_reason() + if reason: + pytest.skip( + f"{request.node.name}: PYEG_LIVE_EGERIA=1 but Egeria at " + f"{settings['platform_url']} is not usable ({reason})" + ) + + with contextlib.ExitStack() as stack: + yield EgeriaBackend(live=live, stack=stack, **settings) + + +@pytest.fixture +def live_my_profile(backend): + """A live MyProfile client, or skip. Used to discover real test inputs.""" + if not backend.live: + pytest.skip("live-only fixture") + from pyegeria import MyProfile + + client = MyProfile(backend.view_server, backend.platform_url, backend.user_id, backend.user_pwd) + client.create_egeria_bearer_token(backend.user_id, backend.user_pwd) + return client + + +@pytest.fixture +def live_team_role_name(backend, live_my_profile): + """A real Team{Leader,Member} role name from the live profile. + + ``find_team_members`` splits the role name on '::' and searches on + everything after the first segment, so this has to be a genuine role name + from the server rather than a synthetic one. + """ + profile = live_my_profile.get_my_profile(report_spec="My-User-MD", output_format="DICT") + roles = (profile or [{}])[0].get("Roles") or [] + for role in roles: + name = role.get("Name") or role.get("Role Name") or "" + if "TeamLeader" in name or "TeamMember" in name: + return name + pytest.skip(f"live profile for {backend.user_id} has no TeamLeader/TeamMember role") + + +def _live_curation_client(backend): + from pyegeria import AutomatedCuration + + client = AutomatedCuration(backend.view_server, backend.platform_url, backend.user_id, backend.user_pwd) + client.create_egeria_bearer_token(backend.user_id, backend.user_pwd) + return client + + +def _live_tech_type_names(client): + """Every technology type display name on the live server, depth-first.""" + names: list[str] = [] + + def walk(node): + if isinstance(node, dict): + name = node.get("displayName") + if name and name != "Root Technology Type": + names.append(name) + for child in node.get("subTypes") or []: + walk(child) + elif isinstance(node, list): + for child in node: + walk(child) + + walk(client.get_tech_type_hierarchy(filter_string="*")) + return names + + +@pytest.fixture +def live_tech_type_name(backend): + """A real technology-type name that resolves to detail on the live server. + + ``tech_type_callback`` passes the screen's selection straight to + ``get_tech_type_detail(filter_string=...)``, which matches on display name, + so the live input has to be a genuine name off the type hierarchy. + """ + if not backend.live: + pytest.skip("live-only fixture") + client = _live_curation_client(backend) + for name in _live_tech_type_names(client): + detail = client.get_tech_type_detail(filter_string=name, output_format="JSON") + if isinstance(detail, dict) and detail.get("displayName"): + return name + pytest.skip("live server has no technology type with retrievable detail") + + +@pytest.fixture +def live_catalog_template(backend): + """A real catalog template entry, exactly as the templates screen passes it. + + `TechnologyTypeTemplatesScreen` hands `tech_type_templates_callback` the raw + `catalogTemplates` entry from `get_tech_type_detail`, so this returns the + same unmodified dict. Most technology types carry no catalog template, so + this scans the hierarchy for one that does. + """ + if not backend.live: + pytest.skip("live-only fixture") + client = _live_curation_client(backend) + for name in _live_tech_type_names(client): + detail = client.get_tech_type_detail(filter_string=name, output_format="JSON") + for template in (detail or {}).get("catalogTemplates") or []: + if template.get("templateGUID"): + return template + pytest.skip("live server has no technology type with a catalog template") + + +@pytest.fixture +def live_template_placeholders(backend, live_catalog_template): + """Placeholder values for a real template, keyed the way the screen keys them. + + `TechnologyTypeTemplatesScreen` builds one Input per placeholder with id + `{name-with-spaces-as-underscores}_placeholder_input`, and the handler + reverses that to recover the placeholder name. Values come from each + placeholder's own `example` so they stay type-valid, with a unique suffix on + identity-ish fields so repeated live runs don't collide on qualifiedName. + """ + suffix = uuid.uuid4().hex[:8] + placeholders = (live_catalog_template.get("specification") or {}).get("placeholderProperty") or [] + + values = {} + for placeholder in placeholders: + if placeholder.get("class") != "PlaceholderProperty": + continue + name = placeholder.get("name") + if not name: + continue + example = placeholder.get("example") or "" + identity_like = any(token in name.lower() for token in ("name", "identifier")) + value = f"pytest-{name}-{suffix}" if identity_like else example + values[f"{name.replace(' ', '_')}_placeholder_input"] = value + return values + @pytest.fixture def sample_profile_data(): diff --git a/tests/micro-tests/my_profile/egeria_backend.py b/tests/micro-tests/my_profile/egeria_backend.py new file mode 100644 index 00000000..b0da6271 --- /dev/null +++ b/tests/micro-tests/my_profile/egeria_backend.py @@ -0,0 +1,255 @@ +""" + PDX-License-Identifier: Apache-2.0 + Copyright Contributors to the ODPi Egeria project. + + Fake/live backend switch for the My Profile App test suite. + + By default every test in this folder runs against in-memory fakes, exactly as + it always has. Setting ``PYEG_LIVE_EGERIA=1`` flips the suite over to a real + Egeria view server (``https://localhost:9443`` by default) so the same tests + exercise the real SDK calls, URLs and response shapes. + + The mechanism is deliberately small: ``EgeriaBackend.patch()`` replaces a + symbol with a plain ``MagicMock`` in fake mode, and with + ``MagicMock(wraps=)`` in live mode. Because a wrapping mock + still records every call, assertions about *how* the app called Egeria + (verbs, report-spec names, params) hold identically in both modes — only + assertions about the returned *data* need to differ, and those go through + :meth:`EgeriaBackend.expect`. + + WARNING: live mode is not read-only. ``create_profile``, + ``create_digital_subscription`` and ``initiate_gov_action_process`` are + executed for real against the configured server. +""" + +from __future__ import annotations + +import contextlib +import importlib +import os +import uuid +from typing import Any, Callable +from unittest.mock import MagicMock, patch + +# Enable live mode with PYEG_LIVE_EGERIA=1 (same switch the rest of the repo uses). +LIVE_ENV = "PYEG_LIVE_EGERIA" + +# Live connection details, each overridable by its own environment variable. +CONNECTION_ENV: dict[str, tuple[str, str]] = { + "platform_url": ("PYEG_PLATFORM_URL", "https://localhost:9443"), + "view_server": ("PYEG_SERVER_NAME", "qs-view-server"), + "user_id": ("PYEG_USER_ID", "garygeeke"), + "user_pwd": ("PYEG_USER_PWD", "secret"), +} + +# pyegeria settings keys that must point at the live server before the app's +# load_app_config() caches them. +PYEGERIA_ENV_FOR = { + "platform_url": "EGERIA_PLATFORM_URL", + "view_server": "EGERIA_VIEW_SERVER", + "user_id": "EGERIA_USER", + "user_pwd": "EGERIA_USER_PASSWORD", +} + + +def env_truthy(name: str, default: bool = False) -> bool: + """Interpret an environment variable as a boolean flag.""" + val = os.getenv(name) + if val is None: + return default + return val.strip().lower() in {"1", "true", "yes", "on"} + + +def live_requested() -> bool: + """True when the caller asked for live Egeria tests.""" + return env_truthy(LIVE_ENV) + + +def connection_settings() -> dict[str, str]: + """Resolve the Egeria connection details, honouring per-field overrides.""" + return {key: os.getenv(env_var, default) for key, (env_var, default) in CONNECTION_ENV.items()} + + +def export_pyegeria_env(settings: dict[str, str] | None = None) -> None: + """Point pyegeria's own config at the live server. + + ``load_app_config()`` caches on first call, so this has to happen before any + test constructs a screen or an app. + """ + settings = settings or connection_settings() + for key, env_var in PYEGERIA_ENV_FOR.items(): + os.environ[env_var] = settings[key] + + +_PROBE_UNSET = object() +_probe_result: Any = _PROBE_UNSET + + +def live_unavailable_reason() -> str | None: + """Return None when the live server is usable, else why it is not. + + The result is probed once per session: a live run that cannot authenticate + should skip cleanly rather than produce a screenful of connection errors. + """ + global _probe_result + if _probe_result is not _PROBE_UNSET: + return _probe_result # type: ignore[return-value] + + settings = connection_settings() + try: + from pyegeria.core._server_client import ServerClient + + client = ServerClient( + server_name=settings["view_server"], + platform_url=settings["platform_url"], + user_id=settings["user_id"], + user_pwd=settings["user_pwd"], + ) + token = client.create_egeria_bearer_token(settings["user_id"], settings["user_pwd"]) + _probe_result = None if token else "the server returned no bearer token" + except Exception as e: # noqa: BLE001 - any failure means "not usable" + _probe_result = f"{type(e).__name__}: {e}" + return _probe_result # type: ignore[return-value] + + +class UnexpectedNetworkCall(RuntimeError): + """Raised when a fake-mode test reaches a real server.""" + + +def block_network(monkeypatch, test_name: str) -> None: + """Fail fake-mode tests that issue a real HTTP request. + + Without this, a test whose Egeria calls are only partly mocked passes + quietly while depending on a reachable server — and then behaves + differently on a machine that has none. pyegeria talks to Egeria through + httpx, so intercepting its transport catches every route. + """ + import httpx + + def _refuse(self, request, *args, **kwargs): + raise UnexpectedNetworkCall( + f"{test_name} tried to reach {request.url} while running against fakes. " + "Patch the Egeria client/function through the `backend` fixture " + "(backend.patch / backend.always_fake), or mark the test " + "`@pytest.mark.allow_network` if the call is genuinely intended." + ) + + monkeypatch.setattr(httpx.Client, "send", _refuse) + monkeypatch.setattr(httpx.AsyncClient, "send", _refuse) + + +def _resolve(target: str) -> Any: + """Resolve a 'module.attribute' patch target to the live object.""" + module_name, _, attr = target.rpartition(".") + if not module_name: + raise ValueError(f"patch target must be 'module.attribute', got {target!r}") + return getattr(importlib.import_module(module_name), attr) + + +# --- live-mode predicates for EgeriaBackend.expect --------------------------- + +def nonempty_str(value: Any) -> bool: + """Live check: a real, non-blank string came back.""" + return isinstance(value, str) and value.strip() != "" + + +def is_int(value: Any) -> bool: + """Live check: an integer came back (karma points, counts, ...).""" + return isinstance(value, int) and not isinstance(value, bool) + + +def at_least(minimum: int) -> Callable[[Any], bool]: + """Live check: a size/count of at least `minimum`.""" + + def _check(value: Any) -> bool: + try: + size = len(value) + except TypeError: + size = value + return isinstance(size, int) and size >= minimum + + return _check + + +class EgeriaBackend: + """Per-test handle on whichever backend the suite is running against.""" + + def __init__(self, live: bool, stack: contextlib.ExitStack, **settings: str) -> None: + self.live = live + self.platform_url = settings["platform_url"] + self.view_server = settings["view_server"] + self.user_id = settings["user_id"] + self.user_pwd = settings["user_pwd"] + self._stack = stack + + # --- patching ------------------------------------------------------------ + + def patch(self, target: str, *, returns: Any = None, side_effect: Any = None) -> MagicMock: + """Switchable patch of `target`. + + Fake mode: a plain MagicMock returning `returns` (or raising + `side_effect`). Live mode: a MagicMock wrapping the real object, so the + call reaches Egeria and the returned mock still records call args. + """ + if self.live: + mock = MagicMock(wraps=_resolve(target)) + else: + mock = MagicMock(return_value=returns, side_effect=side_effect) + self._stack.enter_context(patch(target, mock)) + return mock + + def always_fake(self, target: str, *, returns: Any = None, side_effect: Any = None) -> MagicMock: + """Patch `target` with a pure fake in both modes. + + For paths a live server cannot be asked to produce on demand — injected + exceptions, deliberately empty result sets — where the test is really + about the app's own error handling. + """ + mock = MagicMock(return_value=returns, side_effect=side_effect) + self._stack.enter_context(patch(target, mock)) + return mock + + def patch_object(self, target: Any, attribute: str, replacement: Any) -> Any: + """Patch an attribute that is not Egeria-related in either mode (UI bits).""" + return self._stack.enter_context(patch.object(target, attribute, replacement)) + + # --- assertions ---------------------------------------------------------- + + def expect( + self, + actual: Any, + *, + fake: Any, + live: Callable[[Any], bool] | None = None, + label: str = "value", + ) -> None: + """Assert on data whose exact value depends on the backend. + + In fake mode `actual` must equal `fake`. In live mode `live` (if given) + must accept `actual`; with no `live` predicate the value is not checked, + because the live server legitimately holds different content. + """ + if self.live: + if live is not None: + assert live(actual), f"live-mode check failed for {label}: {actual!r}" + else: + assert actual == fake, f"{label}: expected {fake!r}, got {actual!r}" + + # --- helpers ------------------------------------------------------------- + + def unique(self, value: str) -> str: + """Make a name unique per live run; unchanged in fake mode. + + Egeria enforces uniqueness on qualifiedName, so a live write test that + reuses a fixed display name passes once and then fails with a 409 on + every subsequent run. Fake mode keeps the literal so its assertions + stay exact. + """ + return f"{value} {uuid.uuid4().hex[:8]}" if self.live else value + + def apply_connection(self, target: Any) -> None: + """Point a test harness app at the active backend.""" + target.user_name = self.user_id + target.user_password = self.user_pwd + target.view_server = self.view_server + target.platform_url = self.platform_url diff --git a/tests/micro-tests/my_profile/test_generic_data_view_screen.py b/tests/micro-tests/my_profile/test_generic_data_view_screen.py index c82aad59..4de2fb42 100644 --- a/tests/micro-tests/my_profile/test_generic_data_view_screen.py +++ b/tests/micro-tests/my_profile/test_generic_data_view_screen.py @@ -5,7 +5,7 @@ Unit tests for GenericDataViewScreen and its data sampling functionality in My Profile App. """ -from unittest.mock import MagicMock, AsyncMock, patch +from unittest.mock import MagicMock, AsyncMock import pytest from textual.app import App from textual.widgets import DataTable @@ -222,7 +222,7 @@ async def test_screen_action_subscribe(self): class DummyShopForDataApp(ShopForDataMixin): """Test harness for shop_for_data_handler sampling actions.""" - def __init__(self): + def __init__(self, backend=None): self.pushed_screens = [] self.log_messages = [] self.shown_main_screen = False @@ -230,6 +230,8 @@ def __init__(self): self.user_password = "secret" self.view_server = "qs-view-server" self.platform_url = "https://127.0.0.1:9443" + if backend is not None: + backend.apply_connection(self) self.widgets = {} def log(self, msg, *args, **kwargs): @@ -257,9 +259,13 @@ def notify(self, msg, *args, **kwargs): class TestShopForDataSamplingIntegration: """Integration tests for launching GenericDataViewScreen from shop_for_data_handler.""" + @pytest.mark.live_capable @pytest.mark.asyncio - async def test_request_to_sample_data_source_launches_screen(self): - app = DummyShopForDataApp() + async def test_request_to_sample_data_source_launches_screen(self, backend): + # Sampling a digital_product_catalog_table row calls Egeria to look for + # a matching tabular data set, so the client has to go through backend. + backend.patch("shop_for_data_handler.Egeria") + app = DummyShopForDataApp(backend) mock_table = MagicMock() mock_table.get_row.return_value = ["Product 1", "Product Description", "DP::Product1"] mock_table.row_count = 1 @@ -274,9 +280,11 @@ async def test_request_to_sample_data_source_launches_screen(self): assert screen.data_element_qualified_name == "DP::Product1" assert cb == app.generic_data_view_callback + @pytest.mark.live_capable @pytest.mark.asyncio - async def test_request_to_sample_data_source_with_passed_row_values(self): - app = DummyShopForDataApp() + async def test_request_to_sample_data_source_with_passed_row_values(self, backend): + backend.patch("shop_for_data_handler.Egeria") + app = DummyShopForDataApp(backend) row_values = ["Direct Product", "Direct Desc", "DP::Direct::1"] await app.request_to_sample_data_source("row1", 0, "digital_product_catalog_table", row_values=row_values) @@ -311,16 +319,24 @@ def test_parse_tabular_data_set_report(self): assert rows[1][0] == "REC-2" assert "Value: Val-2" in rows[1][1] + @pytest.mark.live_capable @pytest.mark.asyncio - @patch("shop_for_data_handler.ProductManager") - async def test_generic_data_view_callback_subscribe(self, mock_pm_cls): - mock_pm = MagicMock() - mock_pm_cls.return_value = mock_pm - - app = DummyShopForDataApp() + async def test_generic_data_view_callback_subscribe(self, backend): + """Calls create_digital_subscription — a real write when running live.""" + mock_pm_cls = backend.patch("shop_for_data_handler.ProductManager") + if not backend.live: + mock_pm = MagicMock() + mock_pm_cls.return_value = mock_pm + + app = DummyShopForDataApp(backend) await app.generic_data_view_callback([211, "Product 1", "DP::Product1"]) - mock_pm.create_digital_subscription.assert_called_once_with("DP::Product1") + if backend.live: + # The real client is built inside the handler, so only the class + # construction is observable through the wrapping mock. + assert mock_pm_cls.called + else: + mock_pm.create_digital_subscription.assert_called_once_with("DP::Product1") @pytest.mark.asyncio async def test_generic_data_view_callback_quit(self): @@ -328,29 +344,37 @@ async def test_generic_data_view_callback_quit(self): await app.generic_data_view_callback(210) assert app.shown_main_screen is True + @pytest.mark.live_capable @pytest.mark.asyncio - @patch("shop_for_data_handler.Egeria") - async def test_request_to_sample_data_source_with_egeria_tabular_data(self, mock_egeria_cls): - mock_egeria = MagicMock() - mock_egeria_cls.return_value = mock_egeria - mock_egeria.find_tabular_data_sets.return_value = [{"GUID": "guid-123"}] - mock_egeria.get_tabular_data_set.return_value = { - "tabularDataSetReport": { - "recordCount": 1, - "tableName": "SampleTabular", - "columnDescriptions": [{"columnName": "Col1"}, {"columnName": "Col2"}], - "dataRecords": {"0": ["V1", "V2"]}, + async def test_request_to_sample_data_source_with_egeria_tabular_data(self, backend): + mock_egeria_cls = backend.patch("shop_for_data_handler.Egeria") + if not backend.live: + mock_egeria = MagicMock() + mock_egeria_cls.return_value = mock_egeria + mock_egeria.find_tabular_data_sets.return_value = [{"GUID": "guid-123"}] + mock_egeria.get_tabular_data_set.return_value = { + "tabularDataSetReport": { + "recordCount": 1, + "tableName": "SampleTabular", + "columnDescriptions": [{"columnName": "Col1"}, {"columnName": "Col2"}], + "dataRecords": {"0": ["V1", "V2"]}, + } } - } - app = DummyShopForDataApp() + app = DummyShopForDataApp(backend) row_values = ["Tabular Prod", "Tabular Desc", "DP::Tabular::1"] await app.request_to_sample_data_source("row1", 0, "digital_product_catalog_table", row_values=row_values) assert len(app.pushed_screens) == 1 screen, cb = app.pushed_screens[0] assert isinstance(screen, GenericDataViewScreen) - assert screen.sample_data["tabularDataSetReport"]["tableName"] == "SampleTabular" + if backend.live: + # The synthetic qualified name matches nothing on a real server, so + # the handler falls through with no sample data; what this asserts + # live is that the real find_tabular_data_sets round-trip worked. + assert mock_egeria_cls.called + else: + assert screen.sample_data["tabularDataSetReport"]["tableName"] == "SampleTabular" @pytest.mark.asyncio async def test_request_to_sample_data_source_placeholder_notifies(self): diff --git a/tests/micro-tests/my_profile/test_my_profile_app.py b/tests/micro-tests/my_profile/test_my_profile_app.py index f3190fb9..fa89969e 100644 --- a/tests/micro-tests/my_profile/test_my_profile_app.py +++ b/tests/micro-tests/my_profile/test_my_profile_app.py @@ -3,6 +3,9 @@ Copyright Contributors to the ODPi Egeria project. Full lifecycle, user functionality, and regression tests for MyProfileApp. + + Tests marked `live_capable` run against fakes by default and against a real + Egeria view server when PYEG_LIVE_EGERIA=1 is set. """ from unittest.mock import MagicMock, AsyncMock, patch, PropertyMock @@ -15,9 +18,32 @@ from CreateProfileScreen import CreateProfileScreen from UserIdentitiesScreen import UserIdentitiesScreen from EditElementsScreens import EditProfileScreen +from egeria_backend import at_least, is_int, nonempty_str from pyegeria import PyegeriaException +def stub_profile_client(backend, profile_data, identities, todos): + """Keep MyProfileApp's on_mount off the network in fake mode. + + Any test that enters `app.run_test()` triggers on_mount -> + _load_or_create_profile, which builds a real MyProfile client. Without this + the test silently depends on a reachable server even when the behaviour + under test is mocked. Live mode leaves the real client in place. + """ + if backend.live: + return backend.patch("my_profile_app.MyProfile") + + mock_mp = MagicMock() + mock_mp.create_egeria_bearer_token.return_value = "token" + mock_mp._async_get_my_profile = AsyncMock(return_value=profile_data) + mock_mp.get_my_profile.side_effect = [ + profile_data, # for get_my_profile in new_profile_return + identities, # for User-Identities lookup + ] + mock_mp.get_my_to_dos.return_value = todos + return backend.always_fake("my_profile_app.MyProfile", returns=mock_mp) + + class TestMyProfileAppLifecycle: """Tests for MyProfileApp lifecycle, initialization, and data loading.""" @@ -33,46 +59,48 @@ def test_app_initialization(self): assert app.todos == [] assert app.karma_points == 0 + @pytest.mark.live_capable @pytest.mark.asyncio - @patch("my_profile_app.MyProfile") - async def test_app_on_mount_success(self, mock_mp_cls, sample_profile_data, sample_user_identities, sample_todos_data): - mock_mp = MagicMock() - mock_mp.create_egeria_bearer_token.return_value = "token" - mock_mp._async_get_my_profile = AsyncMock(return_value=sample_profile_data) - mock_mp.get_my_profile.side_effect = [ - sample_profile_data, # for get_my_profile in new_profile_return - sample_user_identities, # for User-Identities lookup - ] - mock_mp.get_my_to_dos.return_value = sample_todos_data - mock_mp_cls.return_value = mock_mp + async def test_app_on_mount_success( + self, backend, sample_profile_data, sample_user_identities, sample_todos_data + ): + # Live mode uses the real MyProfile against the configured view server. + stub_profile_client( + backend, sample_profile_data, sample_user_identities, sample_todos_data + ) app = MyProfileApp() async with app.run_test() as pilot: await pilot.pause() - assert app.karma_points == 150 - assert len(app.projects) == 1 - assert len(app.teams) == 1 - assert len(app.roles) == 1 - assert len(app.todos) == 1 - assert app.user_GUID == "profile-guid-12345" + backend.expect(app.karma_points, fake=150, live=is_int, label="karma_points") + backend.expect(len(app.projects), fake=1, live=at_least(0), label="projects") + backend.expect(len(app.teams), fake=1, live=at_least(0), label="teams") + backend.expect(len(app.roles), fake=1, live=at_least(1), label="roles") + backend.expect(len(app.todos), fake=1, live=at_least(0), label="todos") + backend.expect( + app.user_GUID, fake="profile-guid-12345", live=nonempty_str, label="user_GUID" + ) main_screen = app.get_screen("main") - roles_table = main_screen.query_one("#roles_table", DataTable) - assert roles_table.row_count == 1 - teams_table = main_screen.query_one("#teams_table", DataTable) - assert teams_table.row_count == 1 - todos_table = main_screen.query_one("#todos_table", DataTable) - assert todos_table.row_count == 1 + for table_id, fake_rows in ( + ("#roles_table", 1), + ("#teams_table", 1), + ("#todos_table", 1), + ): + table = main_screen.query_one(table_id, DataTable) + backend.expect( + table.row_count, fake=fake_rows, live=at_least(0), label=table_id + ) blogs_table = main_screen.query_one("#blogs_table", DataTable) assert blogs_table.row_count >= 1 @pytest.mark.asyncio - @patch("my_profile_app.MyProfile") - async def test_app_on_mount_prompt_create_profile(self, mock_mp_cls): + async def test_app_on_mount_prompt_create_profile(self, backend): + # Always faked: a live server cannot be asked for a user with no profile. mock_mp = MagicMock() mock_mp.create_egeria_bearer_token.return_value = "token" mock_mp._async_get_my_profile = AsyncMock(return_value=[]) - mock_mp_cls.return_value = mock_mp + backend.always_fake("my_profile_app.MyProfile", returns=mock_mp) app = MyProfileApp() async with app.run_test() as pilot: @@ -80,12 +108,12 @@ async def test_app_on_mount_prompt_create_profile(self, mock_mp_cls): assert isinstance(app.screen, CreateProfileScreen) @pytest.mark.asyncio - @patch("my_profile_app.MyProfile") - async def test_app_load_profile_exception_exits_402(self, mock_mp_cls): + async def test_app_load_profile_exception_exits_402(self, backend): + # Always faked: exercises the app's error handling, not the server's. mock_mp = MagicMock() mock_mp.create_egeria_bearer_token.return_value = "token" mock_mp._async_get_my_profile = AsyncMock(side_effect=PyegeriaException("Server error")) - mock_mp_cls.return_value = mock_mp + backend.always_fake("my_profile_app.MyProfile", returns=mock_mp) app = MyProfileApp() app.exit = MagicMock() @@ -242,30 +270,50 @@ def test_utility_delegation_wrappers(self): res = app.extract_glossary_terms("GlossaryTerm::TermA, other") assert res == ["TermA"] + @pytest.mark.live_capable @pytest.mark.asyncio - @patch("my_profile_app.exec_report_spec") - async def test_get_data_product_catalog_table_success(self, mock_exec): - mock_exec.return_value = { - "kind": "data", - "data": [ - { - "Display Name": "Catalog 1", - "Description": "Desc 1", - "Qualified Name": "Cat::1", - } - ], - } + async def test_get_data_product_catalog_table_success( + self, backend, sample_profile_data, sample_user_identities, sample_todos_data + ): + stub_profile_client( + backend, sample_profile_data, sample_user_identities, sample_todos_data + ) + mock_exec = backend.patch( + "my_profile_app.exec_report_spec", + returns={ + "kind": "data", + "data": [ + { + "Display Name": "Catalog 1", + "Description": "Desc 1", + "Qualified Name": "Cat::1", + } + ], + }, + ) app = MyProfileApp() async with app.run_test(): rc = app.get_data_product_catalog_table() assert rc == 200 + assert mock_exec.called assert app.digital_product_catalog_table is not None - assert app.digital_product_catalog_table.row_count == 1 + backend.expect( + app.digital_product_catalog_table.row_count, + fake=1, + live=at_least(1), + label="catalog rows", + ) @pytest.mark.asyncio - @patch("my_profile_app.exec_report_spec") - async def test_get_data_product_catalog_table_empty(self, mock_exec): - mock_exec.return_value = {"kind": "empty", "data": []} + async def test_get_data_product_catalog_table_empty( + self, backend, sample_profile_data, sample_user_identities, sample_todos_data + ): + # Always faked: an empty catalog is a server-state the live instance + # cannot be asked to produce. + stub_profile_client( + backend, sample_profile_data, sample_user_identities, sample_todos_data + ) + backend.always_fake("my_profile_app.exec_report_spec", returns={"kind": "empty", "data": []}) app = MyProfileApp() async with app.run_test(): rc = app.get_data_product_catalog_table() diff --git a/tests/micro-tests/my_profile/test_screens.py b/tests/micro-tests/my_profile/test_screens.py index b98cc09c..897f828f 100644 --- a/tests/micro-tests/my_profile/test_screens.py +++ b/tests/micro-tests/my_profile/test_screens.py @@ -5,6 +5,7 @@ Unit tests for My Profile UI modal screens. """ +import uuid from unittest.mock import MagicMock, AsyncMock, patch import pytest from textual.app import App, ComposeResult @@ -53,7 +54,7 @@ class ScreenTestHostApp(App): """Host Textual App with main screen mounted.""" - def __init__(self, screen_factory): + def __init__(self, screen_factory, backend=None): super().__init__() self.screen_factory = screen_factory self.target_screen = None @@ -62,8 +63,10 @@ def __init__(self, screen_factory): self.platform_url = "https://127.0.0.1:9443" self.user_name = "garygeeke" self.user_password = "secret" - self.user = "garygeeke" - self.password = "secret" + if backend is not None: + backend.apply_connection(self) + self.user = self.user_name + self.password = self.user_password self.karma_points = 150 async def on_mount(self): @@ -167,15 +170,23 @@ async def test_search_for_term_screen_actions(self): await pilot.pause() assert app.dismissed_result == 200 + @pytest.mark.live_capable @pytest.mark.asyncio - @patch("SearchForTermScreen.exec_report_spec") - async def test_search_for_term_screen_search(self, mock_exec): - mock_exec.return_value = { - "kind": "text", - "mimeType": "text/markdown", - "content": "## Term Details\nDescription of clinical trial", - } - app = ScreenTestHostApp(lambda: SearchForTermScreen("garygeeke", "secret", "qs-view-server", "https://127.0.0.1:9443")) + async def test_search_for_term_screen_search(self, backend): + backend.patch( + "SearchForTermScreen.exec_report_spec", + returns={ + "kind": "text", + "mimeType": "text/markdown", + "content": "## Term Details\nDescription of clinical trial", + }, + ) + app = ScreenTestHostApp( + lambda: SearchForTermScreen( + backend.user_id, backend.user_pwd, backend.view_server, backend.platform_url + ), + backend, + ) async with app.run_test() as pilot: inp = app.target_screen.query_one("#search_term_input", Input) inp.value = "Clinical" @@ -196,16 +207,59 @@ async def test_create_profile_screen_actions(self): await pilot.pause() assert app.dismissed_result == 200 + @pytest.mark.live_capable @pytest.mark.asyncio - @patch("CreateProfileScreen.MyProfile") - async def test_create_profile_screen_create_profile(self, mock_mp_cls): - mock_mp = MagicMock() - mock_mp.create_egeria_bearer_token.return_value = "token" - mock_mp.add_my_profile.return_value = "profile-guid-999" - mock_mp_cls.return_value = mock_mp - - app = ScreenTestHostApp(lambda: CreateProfileScreen("garygeeke", "secret", "qs-view-server", "https://127.0.0.1:9443")) + async def test_create_profile_screen_create_profile(self, backend): + """Calls add_my_profile — a real write when running live.""" + if backend.live: + # add_my_profile creates the profile for the *calling* user, and the + # server rejects a second one (OMVS-MY-PROFILE-400-001). This path + # is only reachable live for a user who has no profile yet. + from pyegeria import MyProfile + + probe = MyProfile( + backend.view_server, backend.platform_url, backend.user_id, backend.user_pwd + ) + probe.create_egeria_bearer_token(backend.user_id, backend.user_pwd) + if probe.get_my_profile(output_format="DICT", report_spec="My-User-MD"): + pytest.skip( + f"user {backend.user_id} already has a profile; " + "add_my_profile cannot be exercised live" + ) + backend.patch("CreateProfileScreen.MyProfile") + else: + mock_mp = MagicMock() + mock_mp.create_egeria_bearer_token.return_value = "token" + mock_mp.add_my_profile.return_value = "profile-guid-999" + backend.always_fake("CreateProfileScreen.MyProfile", returns=mock_mp) + + app = ScreenTestHostApp( + lambda: CreateProfileScreen( + backend.user_id, backend.user_pwd, backend.view_server, backend.platform_url + ), + backend, + ) async with app.run_test() as pilot: + if backend.live: + # An empty form yields qualifiedName "Person", which the server + # rejects. Fill it with values unique to this run. + suffix = uuid.uuid4().hex[:8] + for field_id, value in { + "#user_employee_id": f"TEST-{suffix}", + "#user_resident_country": "United Kingdom", + "#user_given_names": "Pytest", + "#user_family_name": f"Fixture{suffix}", + "#user_preferred_name": f"Pytest Fixture {suffix}", + "#user_title": "Dr", + "#user_pronouns": "they/them", + "#user_job_title": "Automated test profile", + "#user_description": "Created by the My Profile live test suite", + "#user_preferred_language": "English", + "#user_time_zone": "Europe/London", + }.items(): + app.target_screen.query_one(field_id, Input).value = value + await pilot.pause() + app.target_screen.create_profile() await pilot.pause() assert app.dismissed_result == 200 diff --git a/tests/micro-tests/my_profile/test_shop_for_data_handler.py b/tests/micro-tests/my_profile/test_shop_for_data_handler.py index ab2f9f4e..4cee85da 100644 --- a/tests/micro-tests/my_profile/test_shop_for_data_handler.py +++ b/tests/micro-tests/my_profile/test_shop_for_data_handler.py @@ -3,6 +3,9 @@ Copyright Contributors to the ODPi Egeria project. Unit tests for shop_for_data_handler module. + + Tests marked `live_capable` run against fakes by default and against a real + Egeria view server when PYEG_LIVE_EGERIA=1 is set. """ from unittest.mock import MagicMock, AsyncMock, patch @@ -21,7 +24,7 @@ class DummyShopApp(App, ShopForDataMixin): """Test harness implementing ShopForDataMixin.""" - def __init__(self): + def __init__(self, backend=None): super().__init__() self.pushed_screens = [] self.exit_code = None @@ -31,6 +34,8 @@ def __init__(self): self.user_password = "secret" self.view_server = "qs-view-server" self.platform_url = "https://127.0.0.1:9443" + if backend is not None: + backend.apply_connection(self) self.root_collection_table = MagicMock() self.collections = [] self.selected_item = None @@ -74,24 +79,26 @@ def query_one(self, selector, *args, **kwargs): class TestShopForDataMixin: """Tests for ShopForDataMixin methods.""" + @pytest.mark.live_capable @pytest.mark.asyncio @patch("shop_for_data_handler.DataTable") - @patch("shop_for_data_handler.exec_report_spec") - async def test_handle_shop_for_data_option_success(self, mock_exec, mock_table_cls): - mock_table = MagicMock() - mock_table_cls.return_value = mock_table - mock_exec.return_value = { - "kind": "data", - "data": [ - { - "Display Name": "Test Element", - "Description": "Test Desc", - "Qualified Name": "Test::QN", - } - ], - } + async def test_handle_shop_for_data_option_success(self, mock_table_cls, backend): + mock_table_cls.return_value = MagicMock() + mock_exec = backend.patch( + "shop_for_data_handler.exec_report_spec", + returns={ + "kind": "data", + "data": [ + { + "Display Name": "Test Element", + "Description": "Test Desc", + "Qualified Name": "Test::QN", + } + ], + }, + ) - app = DummyShopApp() + app = DummyShopApp(backend) await app.handle_shop_for_data_option() assert len(app.pushed_screens) == 1 @@ -99,7 +106,9 @@ async def test_handle_shop_for_data_option_success(self, mock_exec, mock_table_c assert isinstance(screen, ShopForDataScreen) assert cb == app.shop_for_data_callback - # Test the underlying worker fetch methods directly + # Test the underlying worker fetch methods directly. The call-args + # assertions below hold in both modes: in live mode the patch wraps the + # real exec_report_spec, so calls are recorded *and* reach Egeria. await ShopForDataMixin.get_digital_product_data.__wrapped__(app) catalog_calls = [ call for call in mock_exec.call_args_list @@ -115,13 +124,13 @@ async def test_handle_shop_for_data_option_success(self, mock_exec, mock_table_c @pytest.mark.asyncio @patch("shop_for_data_handler.DataTable") - @patch("shop_for_data_handler.exec_report_spec") - async def test_handle_shop_for_data_option_empty_data(self, mock_exec, mock_table_cls): - mock_table = MagicMock() - mock_table_cls.return_value = mock_table - mock_exec.return_value = {"kind": "empty", "data": []} + async def test_handle_shop_for_data_option_empty_data(self, mock_table_cls, backend): + # Always faked: an empty result set is not something a live server can + # be asked for on demand. + mock_table_cls.return_value = MagicMock() + backend.always_fake("shop_for_data_handler.exec_report_spec", returns={"kind": "empty", "data": []}) - app = DummyShopApp() + app = DummyShopApp(backend) await app.handle_shop_for_data_option() assert len(app.pushed_screens) == 1 @@ -130,13 +139,14 @@ async def test_handle_shop_for_data_option_empty_data(self, mock_exec, mock_tabl @pytest.mark.asyncio @patch("shop_for_data_handler.DataTable") - @patch("shop_for_data_handler.exec_report_spec") - async def test_handle_shop_for_data_option_exception(self, mock_exec, mock_table_cls): - mock_table = MagicMock() - mock_table_cls.return_value = mock_table - mock_exec.side_effect = PyegeriaException("Network Error") + async def test_handle_shop_for_data_option_exception(self, mock_table_cls, backend): + # Always faked: injected failure, exercising the handler's fallback. + mock_table_cls.return_value = MagicMock() + backend.always_fake( + "shop_for_data_handler.exec_report_spec", side_effect=PyegeriaException("Network Error") + ) - app = DummyShopApp() + app = DummyShopApp(backend) await app.handle_shop_for_data_option() res = await ShopForDataMixin.get_glossary_data.__wrapped__(app) assert res == ["No Data", "Returned by Egeria"] @@ -173,90 +183,102 @@ async def test_shop_for_data_callback_sample_data_source(self): res = await app.shop_for_data_callback([212, "row1", 0, "glossary_table"]) assert res == 200 + @pytest.mark.live_capable @pytest.mark.asyncio - @patch("shop_for_data_handler.exec_report_spec") - async def test_shop_for_data_callback_glossary_selection(self, mock_exec): - mock_exec.return_value = { - "kind": "data", - "data": [ - { - "Display Name": "Term 1", - "Description": "Term 1 description", - "Qualified Name": "GlossaryTerm::T1", - } - ], - } + async def test_shop_for_data_callback_glossary_selection(self, backend): + backend.patch( + "shop_for_data_handler.exec_report_spec", + returns={ + "kind": "data", + "data": [ + { + "Display Name": "Term 1", + "Description": "Term 1 description", + "Qualified Name": "GlossaryTerm::T1", + } + ], + }, + ) - app = DummyShopApp() + app = DummyShopApp(backend) await app.shop_for_data_callback(["glossary", "Clinical Glossary", "Glossary::Clinical"]) assert len(app.pushed_screens) == 1 screen, cb = app.pushed_screens[0] assert isinstance(screen, SelectionOverviewScreen) assert cb == app.overview_callback + @pytest.mark.live_capable @pytest.mark.asyncio - @patch("shop_for_data_handler.exec_report_spec") - async def test_shop_for_data_callback_dictionary_selection(self, mock_exec): - mock_exec.return_value = { - "kind": "data", - "data": [ - { - "Display Name": "Dict 1", - "Description": "Dict 1 desc", - "Qualified Name": "Dictionary::D1", - } - ], - } + async def test_shop_for_data_callback_dictionary_selection(self, backend): + backend.patch( + "shop_for_data_handler.exec_report_spec", + returns={ + "kind": "data", + "data": [ + { + "Display Name": "Dict 1", + "Description": "Dict 1 desc", + "Qualified Name": "Dictionary::D1", + } + ], + }, + ) - app = DummyShopApp() + app = DummyShopApp(backend) await app.shop_for_data_callback(["dictionary", "DataDict", "Dict::QN"]) assert len(app.pushed_screens) == 1 screen, cb = app.pushed_screens[0] assert isinstance(screen, SelectionOverviewScreen) + @pytest.mark.live_capable @pytest.mark.asyncio - @patch("shop_for_data_handler.exec_report_spec") - async def test_shop_for_data_callback_domain_selection(self, mock_exec): - mock_exec.return_value = { - "kind": "data", - "data": [ - { - "Display Name": "Domain 1", - "Description": "Domain 1 desc", - "Qualified Name": "Domain::Dom1", - } - ], - } + async def test_shop_for_data_callback_domain_selection(self, backend): + backend.patch( + "shop_for_data_handler.exec_report_spec", + returns={ + "kind": "data", + "data": [ + { + "Display Name": "Domain 1", + "Description": "Domain 1 desc", + "Qualified Name": "Domain::Dom1", + } + ], + }, + ) - app = DummyShopApp() + app = DummyShopApp(backend) await app.shop_for_data_callback(["domain", "FinanceDomain", "Domain::Finance"]) assert len(app.pushed_screens) == 1 screen, cb = app.pushed_screens[0] assert isinstance(screen, SelectionOverviewScreen) + @pytest.mark.live_capable @pytest.mark.asyncio - @patch("shop_for_data_handler.exec_report_spec") - async def test_shop_for_data_callback_catalog_selection(self, mock_exec): - mock_exec.return_value = { - "kind": "data", - "data": [ - { - "Display Name": "Cat 1", - "Description": "Cat 1 desc", - "Qualified Name": "Catalog::C1", - } - ], - } + async def test_shop_for_data_callback_catalog_selection(self, backend): + backend.patch( + "shop_for_data_handler.exec_report_spec", + returns={ + "kind": "data", + "data": [ + { + "Display Name": "Cat 1", + "Description": "Cat 1 desc", + "Qualified Name": "Catalog::C1", + } + ], + }, + ) - app = DummyShopApp() + app = DummyShopApp(backend) await app.shop_for_data_callback(["catalog", "ProductCat", "Catalog::Prod"]) assert len(app.pushed_screens) == 1 screen, cb = app.pushed_screens[0] assert isinstance(screen, SelectionOverviewScreen) @pytest.mark.asyncio - async def test_shop_for_data_callback_collection_selection(self): - app = DummyShopApp() + async def test_shop_for_data_callback_collection_selection(self, backend): + app = DummyShopApp(backend) app.collections = [{"Qualified Name": "Coll::Root", "Containing Members": "Folder1, Folder2"}] await app.shop_for_data_callback(["collection", "Coll::Root", "Root Collection"]) assert len(app.pushed_screens) == 1 @@ -291,27 +313,33 @@ def test_create_subscription_callback_cancelled(self): app.create_subscription_callback(None) assert any("cancelled" in msg for msg in app.log_messages) - def test_create_subscription_callback_success(self): - app = DummyShopApp() - app.create_subscription_callback("Sub-Result-123") + @pytest.mark.live_capable + def test_create_subscription_callback_success(self, backend): + # Reaches ProductManager.create_digital_subscription: a real write when live. + backend.patch("shop_for_data_handler.ProductManager") + app = DummyShopApp(backend) + app.create_subscription_callback(backend.unique("Sub-Result-123")) assert any("Subscription created" in msg for msg in app.log_messages) - def test_create_subscription_callback_dict_with_guid(self): - app = DummyShopApp() + @pytest.mark.live_capable + def test_create_subscription_callback_dict_with_guid(self, backend): + backend.patch("shop_for_data_handler.ProductManager") + app = DummyShopApp(backend) app.selected_item = "fallback-guid" app.create_subscription_callback({ "externalSourceGUID": "item-guid-123", - "displayName": "Test Sub", + "displayName": backend.unique("Test Sub"), "Status": "ACTIVE", "description": "Test Desc", "identifier": "TS1", }) assert any("Subscription created" in msg for msg in app.log_messages) + @pytest.mark.live_capable @pytest.mark.asyncio - @patch("shop_for_data_handler.ProductManager") - async def test_shop_for_data_callback_direct_subscribe(self, mock_pm_cls): - app = DummyShopApp() + async def test_shop_for_data_callback_direct_subscribe(self, backend): + backend.patch("shop_for_data_handler.ProductManager") + app = DummyShopApp(backend) res = await app.shop_for_data_callback([ 211, "row1", @@ -326,10 +354,11 @@ async def test_shop_for_data_callback_direct_subscribe(self, mock_pm_cls): assert cb == app.create_subscription_callback assert app.selected_item == "guid-prod-123" + @pytest.mark.live_capable @pytest.mark.asyncio - @patch("shop_for_data_handler.ProductManager") - async def test_request_to_subscribe_data_source_placeholder(self, mock_pm_cls): - app = DummyShopApp() + async def test_request_to_subscribe_data_source_placeholder(self, backend): + backend.patch("shop_for_data_handler.ProductManager") + app = DummyShopApp(backend) app.handle_shop_for_data_option = AsyncMock() await app.request_to_subscribe_data_source( "row1", @@ -340,29 +369,41 @@ async def test_request_to_subscribe_data_source_placeholder(self, mock_pm_cls): assert any("No valid data element selected to subscribe" in msg for msg in app.log_messages) assert app.handle_shop_for_data_option.called - @patch("shop_for_data_handler.ProductManager") - def test_create_subscription_callback_creates_subscription_with_client(self, mock_pm_cls): - mock_client = MagicMock() - mock_pm_cls.return_value = mock_client - mock_client.create_digital_subscription.return_value = "created-sub-guid" + @pytest.mark.live_capable + def test_create_subscription_callback_creates_subscription_with_client(self, backend): + mock_pm_cls = backend.patch("shop_for_data_handler.ProductManager") + if not backend.live: + mock_client = MagicMock() + mock_pm_cls.return_value = mock_client + mock_client.create_digital_subscription.return_value = "created-sub-guid" - app = DummyShopApp() + app = DummyShopApp(backend) app.selected_item = "guid-prod-123" + display_name = backend.unique("My Sub") app.create_subscription_callback({ - "displayName": "My Sub", + "displayName": display_name, "description": "Sub Desc", "Status": "ACTIVE", "identifier": "MS-01", "externalSourceGUID": "guid-prod-123", }) - assert mock_client.create_digital_subscription.called - call_args = mock_client.create_digital_subscription.call_args[0][0] - assert call_args["class"] == "NewAgreementRequestBody" - assert call_args["initialStatus"] == "ACTIVE" - assert call_args["externalSourceGUID"] == "guid-prod-123" - assert call_args["properties"]["displayName"] == "My Sub" - assert any("Created digital subscription successfully" in msg for msg in app.log_messages) + # The request body is built before the client call, so it is only + # inspectable via the mock in fake mode; live mode instead asserts the + # real create reached Egeria without error. + if backend.live: + assert mock_pm_cls.called + assert any("Created digital subscription successfully" in msg for msg in app.log_messages), ( + f"live create_digital_subscription failed: {app.log_messages}" + ) + else: + assert mock_client.create_digital_subscription.called + call_args = mock_client.create_digital_subscription.call_args[0][0] + assert call_args["class"] == "NewAgreementRequestBody" + assert call_args["initialStatus"] == "ACTIVE" + assert call_args["externalSourceGUID"] == "guid-prod-123" + assert call_args["properties"]["displayName"] == display_name + assert any("Created digital subscription successfully" in msg for msg in app.log_messages) def test_on_worker_state_changed_product_group(self): app = DummyShopApp() diff --git a/tests/micro-tests/my_profile/test_team_roles_handler.py b/tests/micro-tests/my_profile/test_team_roles_handler.py index 75d70289..236c2971 100644 --- a/tests/micro-tests/my_profile/test_team_roles_handler.py +++ b/tests/micro-tests/my_profile/test_team_roles_handler.py @@ -3,23 +3,26 @@ Copyright Contributors to the ODPi Egeria project. Unit tests for team_roles_handler module. + + Tests marked `live_capable` run against fakes by default and against a real + Egeria view server when PYEG_LIVE_EGERIA=1 is set. """ -from unittest.mock import MagicMock, patch +from unittest.mock import MagicMock import pytest -from textual.widgets import DataTable from team_roles_handler import TeamRolesMixin from MyTeamScreen import MyTeam from ShopForDataScreen import ShopForDataScreen from StatusScreen import StatusScreen +from egeria_backend import at_least, nonempty_str from pyegeria import PyegeriaException class DummyTeamRolesApp(TeamRolesMixin): """Test harness implementing TeamRolesMixin.""" - def __init__(self): + def __init__(self, backend=None): self.pushed_screens = [] self.exit_code = None self.shown_main_screen = False @@ -28,6 +31,8 @@ def __init__(self): self.user_password = "secret" self.view_server = "qs-view-server" self.platform_url = "https://127.0.0.1:9443" + if backend is not None: + backend.apply_connection(self) self.team_members = [] self.widgets = {} self.screen = MagicMock() @@ -72,14 +77,24 @@ def test_handle_roles_table_row_selection_non_team(self): assert res == 201 assert len(app.pushed_screens) == 0 - @patch("team_roles_handler.exec_report_spec") - def test_handle_roles_table_row_selection_team_leader(self, mock_exec, sample_team_members_response): - app = DummyTeamRolesApp() - mock_exec.return_value = sample_team_members_response + @pytest.mark.live_capable + def test_handle_roles_table_row_selection_team_leader( + self, backend, sample_team_members_response, request + ): + app = DummyTeamRolesApp(backend) + backend.patch("team_roles_handler.exec_report_spec", returns=sample_team_members_response) + + # Live mode needs a role name the server actually knows: find_team_members + # searches on everything after the first '::' segment. + role_name = ( + request.getfixturevalue("live_team_role_name") + if backend.live + else "Department::101::TeamLeader" + ) mock_event = MagicMock() mock_table = MagicMock() - mock_table.get_row.return_value = ["Department::101::TeamLeader", "TeamLeader", "Desc", "guid1"] + mock_table.get_row.return_value = [role_name, "TeamLeader", "Desc", "guid1"] mock_event.data_table = mock_table mock_event.row_key = "k1" @@ -89,34 +104,47 @@ def test_handle_roles_table_row_selection_team_leader(self, mock_exec, sample_te screen, cb = app.pushed_screens[0] assert isinstance(screen, MyTeam) assert cb == app.my_team_callback - assert len(app.team_members) == 2 - assert app.team_members[0] == ["Gary Geeke", "TeamLeader", "profile-guid-12345"] - - @patch("team_roles_handler.exec_report_spec") - def test_find_team_members_success(self, mock_exec, sample_team_members_response): - app = DummyTeamRolesApp() - mock_exec.return_value = sample_team_members_response - - members, dname, qname, cat, desc = app.find_team_members("Department::101::TeamLeader") - assert len(members) == 2 - assert dname == "IT Infrastructure Team" - assert qname == "Team::IT_Infra" - assert cat == "Operations" - assert desc == "Team responsible for core infrastructure" - - @patch("team_roles_handler.exec_report_spec") - def test_find_team_members_pyegeria_exception(self, mock_exec): - app = DummyTeamRolesApp() - mock_exec.side_effect = PyegeriaException("API Error") + backend.expect(len(app.team_members), fake=2, live=at_least(1), label="team_members") + backend.expect( + app.team_members[0], + fake=["Gary Geeke", "TeamLeader", "profile-guid-12345"], + live=lambda row: isinstance(row, list) and len(row) == 3, + label="first team member", + ) + + @pytest.mark.live_capable + def test_find_team_members_success(self, backend, sample_team_members_response, request): + app = DummyTeamRolesApp(backend) + backend.patch("team_roles_handler.exec_report_spec", returns=sample_team_members_response) + + role_name = ( + request.getfixturevalue("live_team_role_name") + if backend.live + else "Department::101::TeamLeader" + ) + + members, dname, qname, cat, desc = app.find_team_members(role_name) + backend.expect(len(members), fake=2, live=at_least(1), label="members") + backend.expect(dname, fake="IT Infrastructure Team", live=nonempty_str, label="display name") + backend.expect(qname, fake="Team::IT_Infra", live=nonempty_str, label="qualified name") + backend.expect(cat, fake="Operations", label="category") + backend.expect( + desc, fake="Team responsible for core infrastructure", label="description" + ) + + def test_find_team_members_pyegeria_exception(self, backend): + # Always faked: injected failure, exercising the handler's error path. + app = DummyTeamRolesApp(backend) + backend.always_fake("team_roles_handler.exec_report_spec", side_effect=PyegeriaException("API Error")) members, dname, qname, cat, desc = app.find_team_members("Department::101::TeamLeader") assert members == [] assert app.exit_code == 440 - @patch("team_roles_handler.exec_report_spec") - def test_find_team_members_empty_kind(self, mock_exec): - app = DummyTeamRolesApp() - mock_exec.return_value = {"kind": "empty"} + def test_find_team_members_empty_kind(self, backend): + # Always faked: an 'empty' result is not something a live server can be asked for. + app = DummyTeamRolesApp(backend) + backend.always_fake("team_roles_handler.exec_report_spec", returns={"kind": "empty"}) members, dname, qname, cat, desc = app.find_team_members("Department::101::TeamLeader") assert members == [] @@ -124,10 +152,10 @@ def test_find_team_members_empty_kind(self, mock_exec): screen, cb = app.pushed_screens[0] assert isinstance(screen, StatusScreen) - @patch("team_roles_handler.exec_report_spec") - def test_find_team_members_no_members_found(self, mock_exec): - app = DummyTeamRolesApp() - mock_exec.return_value = {"kind": "data", "data": []} + def test_find_team_members_no_members_found(self, backend): + # Always faked: as above, a team with no members is a fixed fake state. + app = DummyTeamRolesApp(backend) + backend.always_fake("team_roles_handler.exec_report_spec", returns={"kind": "data", "data": []}) members, dname, qname, cat, desc = app.find_team_members("Department::101::TeamLeader") assert members == [] @@ -163,30 +191,37 @@ def test_search_for_term_callback_error(self): screen, cb = app.pushed_screens[0] assert isinstance(screen, StatusScreen) - @patch("team_roles_handler.exec_report_spec") - def test_display_glossary_term_details_success(self, mock_exec): - app = DummyTeamRolesApp() - mock_exec.return_value = {"kind": "data", "data": {"displayName": "Test Term"}} + @pytest.mark.live_capable + def test_display_glossary_term_details_success(self, backend): + app = DummyTeamRolesApp(backend) + backend.patch( + "team_roles_handler.exec_report_spec", + returns={"kind": "data", "data": {"displayName": "Test Term"}}, + ) mock_container = MagicMock() app.screen.query_one.return_value = mock_container - ret = app.display_glossary_term_details("Test Term") + # '*' matches whatever terms the live glossary holds; the fake ignores it. + ret = app.display_glossary_term_details("*" if backend.live else "Test Term") assert ret == 200 - mock_container.mount.assert_called_once() + # The handler mounts one Static per field of a dict payload; a live JSON + # payload is a list, which mounts nothing, so only assert on the fake. + if not backend.live: + mock_container.mount.assert_called_once() - @patch("team_roles_handler.exec_report_spec") - def test_display_glossary_term_details_pyegeria_exception(self, mock_exec): - app = DummyTeamRolesApp() - mock_exec.side_effect = PyegeriaException("API Error") + def test_display_glossary_term_details_pyegeria_exception(self, backend): + # Always faked: injected failure, exercising the handler's error path. + app = DummyTeamRolesApp(backend) + backend.always_fake("team_roles_handler.exec_report_spec", side_effect=PyegeriaException("API Error")) ret = app.display_glossary_term_details("Test Term") assert ret == 440 assert app.exit_code == 440 - @patch("team_roles_handler.exec_report_spec") - def test_display_glossary_term_details_empty(self, mock_exec): - app = DummyTeamRolesApp() - mock_exec.return_value = None + def test_display_glossary_term_details_empty(self, backend): + # Always faked: a null response is not a live-server state. + app = DummyTeamRolesApp(backend) + backend.always_fake("team_roles_handler.exec_report_spec", returns=None) ret = app.display_glossary_term_details("Test Term") assert ret == 440 diff --git a/tests/micro-tests/my_profile/test_tech_types_handler.py b/tests/micro-tests/my_profile/test_tech_types_handler.py index 947a5580..a6eb8525 100644 --- a/tests/micro-tests/my_profile/test_tech_types_handler.py +++ b/tests/micro-tests/my_profile/test_tech_types_handler.py @@ -3,9 +3,13 @@ Copyright Contributors to the ODPi Egeria project. Unit tests for tech_types_handler module. + + Tests marked `live_capable` run against fakes by default and against a real + Egeria view server when PYEG_LIVE_EGERIA=1 is set. """ -from unittest.mock import MagicMock, AsyncMock, patch +import re +from unittest.mock import MagicMock, AsyncMock import pytest from tech_types_handler import TechTypesMixin @@ -16,13 +20,14 @@ TechnologyTypeTemplatesScreen, TechnologyTypeProcessesScreen, ) +from egeria_backend import nonempty_str from pyegeria import PyegeriaException class DummyTechTypesApp(TechTypesMixin): """Test harness implementing TechTypesMixin.""" - def __init__(self): + def __init__(self, backend=None): self.pushed_screens = [] self.exit_code = None self.shown_main_screen = False @@ -32,6 +37,8 @@ def __init__(self): self.karma_points = 150 self.view_server = "qs-view-server" self.platform_url = "https://127.0.0.1:9443" + if backend is not None: + backend.apply_connection(self) self.tech_type_response = [] self.tech_type_list = [] self.tech_type_data = {} @@ -61,12 +68,18 @@ def status_callback(self, *args, **kwargs): class TestTechTypesMixin: """Tests for TechTypesMixin methods.""" + @pytest.mark.live_capable @pytest.mark.asyncio - @patch.object(TechTypesMixin, "fetch_technology_types", new_callable=AsyncMock) - async def test_handle_technology_types_option_success(self, mock_fetch): - app = DummyTechTypesApp() - app.tech_type_response = [{"displayName": "Postgres"}] - app.tech_type_list = [{"displayName": "Postgres"}] + async def test_handle_technology_types_option_success(self, backend): + app = DummyTechTypesApp(backend) + if backend.live: + # Let the real hierarchy fetch populate tech_type_response/list. + await app.fetch_technology_types() + assert app.tech_type_response, "live server returned no technology type hierarchy" + else: + backend.patch_object(TechTypesMixin, "fetch_technology_types", AsyncMock()) + app.tech_type_response = [{"displayName": "Postgres"}] + app.tech_type_list = [{"displayName": "Postgres"}] await app.handle_technology_types_option() @@ -76,39 +89,54 @@ async def test_handle_technology_types_option_success(self, mock_fetch): assert cb == app.tech_type_callback @pytest.mark.asyncio - @patch.object(TechTypesMixin, "fetch_technology_types", new_callable=AsyncMock) - async def test_handle_technology_types_option_empty(self, mock_fetch): - app = DummyTechTypesApp() + async def test_handle_technology_types_option_empty(self, backend): + # Always faked: a server with no technology types is a fixed fake state. + app = DummyTechTypesApp(backend) + backend.patch_object(TechTypesMixin, "fetch_technology_types", AsyncMock()) app.tech_type_response = [] await app.handle_technology_types_option() assert app.exit_code == 200 @pytest.mark.asyncio - @patch.object(TechTypesMixin, "fetch_technology_types", new_callable=AsyncMock) - async def test_handle_technology_types_option_error(self, mock_fetch): - app = DummyTechTypesApp() + async def test_handle_technology_types_option_error(self, backend): + # Always faked: injected error response. + app = DummyTechTypesApp(backend) + backend.patch_object(TechTypesMixin, "fetch_technology_types", AsyncMock()) app.tech_type_response = "404" await app.handle_technology_types_option() assert app.exit_code == 404 + @pytest.mark.live_capable @pytest.mark.asyncio - @patch("tech_types_handler.AutomatedCuration") - async def test_tech_type_callback_valid_selection(self, mock_ac_cls, sample_tech_type_detail): - app = DummyTechTypesApp() - mock_ac = MagicMock() - mock_ac.create_egeria_bearer_token.return_value = "token" - mock_ac.get_tech_type_detail.return_value = sample_tech_type_detail - mock_ac_cls.return_value = mock_ac - - res = await app.tech_type_callback("tech-type-guid-999") + async def test_tech_type_callback_valid_selection( + self, backend, sample_tech_type_detail, request + ): + app = DummyTechTypesApp(backend) + if backend.live: + backend.patch("tech_types_handler.AutomatedCuration") + # get_tech_type_detail matches on display name, so use a real one. + selection = request.getfixturevalue("live_tech_type_name") + else: + mock_ac = MagicMock() + mock_ac.create_egeria_bearer_token.return_value = "token" + mock_ac.get_tech_type_detail.return_value = sample_tech_type_detail + backend.always_fake("tech_types_handler.AutomatedCuration", returns=mock_ac) + selection = "tech-type-guid-999" + + res = await app.tech_type_callback(selection) assert res == 200 assert len(app.pushed_screens) == 1 screen, cb = app.pushed_screens[0] assert isinstance(screen, TechnologyTypeOptionsScreen) assert cb == app.tech_type_options_callback - assert app.tech_type_name == "PostgreSQL Database" + backend.expect( + app.tech_type_name, + fake="PostgreSQL Database", + live=nonempty_str, + label="tech_type_name", + ) assert "specificationMermaidGraph" not in app.tech_type_data @pytest.mark.asyncio @@ -124,12 +152,12 @@ async def test_tech_type_callback_cancelled(self): assert app.shown_main_screen is True @pytest.mark.asyncio - @patch("tech_types_handler.AutomatedCuration") - async def test_tech_type_callback_exception(self, mock_ac_cls): - app = DummyTechTypesApp() + async def test_tech_type_callback_exception(self, backend): + # Always faked: injected auth failure, exercising the handler's error path. + app = DummyTechTypesApp(backend) mock_ac = MagicMock() mock_ac.create_egeria_bearer_token.side_effect = PyegeriaException("Auth failed") - mock_ac_cls.return_value = mock_ac + backend.always_fake("tech_types_handler.AutomatedCuration", returns=mock_ac) await app.tech_type_callback("tech-type-guid-999") assert len(app.pushed_screens) == 2 # StatusScreen pushed on error, then TechnologyTypeOptionsScreen @@ -203,28 +231,44 @@ def test_unpack_egeria_data_invalid(self): app.tech_type_data = 12345 assert app.unpack_egeria_data() == 417 - @patch("tech_types_handler.AutomatedCuration") - def test_tech_type_templates_callback_success(self, mock_ac_cls): - app = DummyTechTypesApp() - app.autoc = MagicMock() - mock_instance = MagicMock() - mock_instance.initiate_gov_action_process.return_value = "new-proc-guid-999" - mock_ac_cls.return_value = mock_instance - - input_result = [ - "input", - {"database_name_placeholder_input": "mydb"}, - { - "Catalog Template GUID": "templ-guid-1", - "typeName": "Database", - }, - ] - app.tech_type_templates_callback(input_result) + @pytest.mark.live_capable + def test_tech_type_templates_callback_success(self, backend, request): + """Creates an element from a catalog template — a real write when live.""" + app = DummyTechTypesApp(backend) + + if backend.live: + from pyegeria import AutomatedCuration + + app.autoc = AutomatedCuration( + app.view_server, app.platform_url, app.user_name, app.user_password + ) + backend.patch("tech_types_handler.AutomatedCuration") + full_template = request.getfixturevalue("live_catalog_template") + placeholders = request.getfixturevalue("live_template_placeholders") + else: + app.autoc = MagicMock() + mock_instance = MagicMock() + mock_instance.create_elem_from_template.return_value = "new-elem-guid-999" + backend.always_fake("tech_types_handler.AutomatedCuration", returns=mock_instance) + full_template = {"templateGUID": "templ-guid-1"} + placeholders = {"database_name_placeholder_input": "mydb"} + + input_result = ["input", placeholders, full_template] + rc = app.tech_type_templates_callback(input_result) assert len(app.pushed_screens) == 1 screen, cb = app.pushed_screens[0] assert isinstance(screen, StatusScreen) - assert "new-proc-guid-999" in screen.status_message + if backend.live: + # 420 is the handler's "create failed" return. + assert rc != 420, f"live create_elem_from_template failed: {screen.status_message}" + # The handler keeps the new GUID local, quoting it into the status + # message, so that is where the test reads it back from. + quoted = re.search(r"'([^']+)'", screen.status_message) + assert quoted, f"no GUID in status message: {screen.status_message}" + assert nonempty_str(quoted.group(1)) and quoted.group(1) != "None" + else: + assert "new-elem-guid-999" in screen.status_message def test_tech_type_templates_callback_invalid(self): app = DummyTechTypesApp()