Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
12 changes: 6 additions & 6 deletions my_egeria/my_egeria/DemoCode/My_Profile/MainScreen.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
20 changes: 10 additions & 10 deletions my_egeria/my_egeria/DemoCode/My_Profile/my_profile_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -468,63 +468,63 @@ 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", "")),
str(r.get("Type", "")),
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", "")),
str(t.get("Team Name", "")),
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", "")),
str(b.get("time", "")),
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", "")),
str(j.get("time", "")),
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", "")),
str(td.get("Activity Status", "")),
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", "")),
str(ui.get("User ID", "")),
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", "")),
str(c.get("Name", "")),
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)
Expand All @@ -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":
Expand Down
17 changes: 5 additions & 12 deletions my_egeria/my_egeria/DemoCode/My_Profile/shop_for_data_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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:
Expand Down
10 changes: 6 additions & 4 deletions my_egeria/my_egeria/DemoCode/My_Profile/tech_types_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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):
Expand Down
149 changes: 149 additions & 0 deletions tests/micro-tests/my_profile/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,149 @@
<!-- SPDX-License-Identifier: CC-BY-4.0 -->
<!-- Copyright Contributors to the ODPi Egeria project. -->

# 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=<the real object>)`, 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=<predicate>)`: 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=<fake payload>)`.
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.
Loading
Loading