diff --git a/docs/My-Egeria-Doc.md b/docs/My-Egeria-Doc.md
index bf1f2ec6..25584095 100644
--- a/docs/My-Egeria-Doc.md
+++ b/docs/My-Egeria-Doc.md
@@ -15,7 +15,9 @@ All source files named here are in `my_egeria/my_egeria/DemoCode/My_Profile/`.
| :--- | :--- |
| `my_egeria.py` | `MyProfileApp`: startup, profile loading, table population, menu dispatch, comments-from-main |
| `MainScreen.py` | The dashboard: tables, Other Functions menu, row-selection tracking |
-| `SplashScreen.py` | Welcome card and Change User |
+| `SplashScreen.py` | Welcome card; its Change User button opens `ChangeUserScreen` |
+| `ChangeUserScreen.py` | Collects a new user and password and checks Egeria accepts them |
+| `ConnectionErrorScreen.py` | Shown when the profile can't be loaded; retries until Egeria accepts the credentials |
| `CreateProfileScreen.py` | First-time profile creation |
| `elements_crud_handler.py` | `ElementsCrudMixin`: add/edit/delete routing, comments screen routing |
| `AddToElementsScreens.py` | `BaseAddScreen` and every Add screen |
@@ -31,7 +33,7 @@ All source files named here are in `my_egeria/my_egeria/DemoCode/My_Profile/`.
| `network_handler.py`, `MyNetworkScreen.py` | `NetworkMixin` (peers linked by `Peer` relationships), `MyNetworkScreen` and `ConfirmRemovePeerScreen` |
| `UserIdentitiesScreen.py` | Read-only user identity table |
| `StatusScreen.py` | Result message with copy-GUID |
-| `profile_utils.py` | Environment check and data clean-up helpers |
+| `profile_utils.py` | Environment check, Egeria access check (`check_egeria_access`) and data clean-up helpers |
| `my_profile.tcss` | Layout and theme |
| `RETURN_CODES.md`, `catalog_api_calls.md` | Return-code list and Shop for Data API notes |
@@ -47,7 +49,10 @@ classDiagram
BINDINGS: q, r, ctrl+f, f3
on_mount()
mainline(splash_return)
+ change_user() / change_user_callback()
+ connection_error_callback()
_load_profile_and_populate()
+ _after_profile_load()
_process_profile_data()
_populate_tables()
handle_option_selected()
@@ -103,15 +108,23 @@ flowchart TD
PUSH_MAIN --> TASK["start task: _load_profile_and_populate()"]
PUSH_MAIN --> SPLASH["push_screen('splash', callback=mainline)"]
SPLASH -- "Continue to App: dismiss()" --> MAINLINE{"mainline(result)"}
- SPLASH -- "Change User: dismiss([user, pwd])" --> MAINLINE
- MAINLINE -- "list (user changed)" --> RELOAD["cancel task, set user,
refresh_bindings(),
_load_or_create_profile()"]
+ SPLASH -- "Change User → ChangeUserScreen:
dismiss([user, pwd])" --> MAINLINE
+ MAINLINE -- "list (user changed)" --> RELOAD["cancel task,
change_user(user, pwd)"]
MAINLINE -- "None" --> AWAIT["await task"]
TASK --> GET["MyProfile._async_get_my_profile('My-User-MD', DICT)"]
- GET -- "PyegeriaException" --> EXIT402(["exit(402)"])
+ GET -- "PyegeriaConnectionException" --> ERR["set load_error,
return False"]
+ GET -- "PyegeriaException" --> ACCESS["check_egeria_access()
for the reason"] --> ERR
GET --> PROC["_process_profile_data()
• get_my_profile('User-Identities')
• get_my_to_dos('My-User-ToDos')
• resolve user GUID (fallback: 'Actor-Profiles')"]
PROC --> POP["_populate_tables()"]
- AWAIT --> FOUND{"Profile found?"}
- RELOAD --> FOUND
+ AWAIT --> AFTER{"_after_profile_load()"}
+ RELOAD --> AFTER
+ AFTER -- "load_error set" --> CES["ConnectionErrorScreen(load_error)
callback=connection_error_callback"]
+ CES -- "200 (Egeria accepts credentials)" --> RELOAD2["_load_or_create_profile()"] --> AFTER
+ CES -- "201 Change User" --> CUS["ChangeUserScreen
callback=change_user_callback"]
+ CUS -- "[user, pwd]" --> RELOAD
+ CUS -- "200 (cancelled)" --> AFTER
+ CES -- "402 Quit" --> EXIT402(["exit(402)"])
+ AFTER -- "no error" --> FOUND{"Profile found?"}
FOUND -- "Yes" --> DASH["Main dashboard"]
FOUND -- "No" --> CPS["CreateProfileScreen
callback=new_profile_return"]
CPS -- "200" --> REREAD["re-read 'My-User-MD', populate"] --> DASH
@@ -120,8 +133,12 @@ flowchart TD
- **Parallel loading.** The profile load starts as soon as the app mounts, while the splash screen is still showing. If the user switches account, the background task is cancelled and the load is repeated for the new user.
- **Escape on the splash screen** dismisses it, the same as **Continue to App**. It must dismiss rather than `pop_screen`, because `pop_screen` discards the pushed screen's callback, so `mainline` (and the missing-profile check) would never run.
-- **Change User** also writes the new user and password into pyegeria's cached `settings.User_Profile`. Screens read the user from there, so every screen opened afterwards runs its Egeria calls as the new user.
-- **Change User check.** The splash screen builds an `Egeria` client from the new credentials and closes it again, but it does not request a token. Bad credentials are therefore only found when the profile load fails.
+- **Load failures don't exit.** `_load_profile_and_populate` catches the error, records a message in `self.load_error` and returns `False`. `_after_profile_load` (shared by the startup path and `_load_or_create_profile`) checks `load_error` *before* the missing-profile check, so an unreachable Egeria never sends the user to Create Profile. A `PyegeriaConnectionException` gives a fixed "Unable to connect to the Egeria platform at …" message; any other `PyegeriaException` calls `check_egeria_access` for the reason, because `str(e)` is often empty (for example on a 401).
+- **`ConnectionErrorScreen`** retries every `RETRY_INTERVAL` (10) seconds from a 1-second `set_interval` countdown, or on **Retry Now** (`r`). Each attempt calls `check_egeria_access` with the user in `settings.User_Profile`, and updates the reason shown if it changed. It dismisses with `200` once Egeria accepts the credentials, `201` for **Change User** (`u`) and `402` for **Quit** (`q`), stopping the timer each time. `connection_error_callback` reloads the profile on `200`, pushes `ChangeUserScreen` on `201`, and otherwise calls `exit(402)`.
+- **`check_egeria_access(view_server, platform_url, user, password)`** returns `None` if Egeria is reachable and accepts the credentials, otherwise a message. The `Egeria` constructor checks the platform answers; the token request then checks the credentials. A successful `create_egeria_bearer_token()` is not proof on its own: on a 401 it returns Egeria's JSON error body as if it were the token, so a token starting with `{` is treated as a rejection and its `exceptionErrorMessage` is reported.
+- **Change User** is reached three ways: the splash screen's button, the **Change User** menu option, and the connection error screen. All three push `ChangeUserScreen`, which validates the new credentials with `check_egeria_access` before dismissing with `[user, password]`; on a failure it shows the reason and stays open. It dismisses with `200` on Cancel or `Escape`. From the splash screen, a cancel leaves the splash showing; a list is passed on through the splash's own `dismiss` to `mainline`.
+- **`change_user(user, pwd)`** does the switch. It sets the app's user and password, writes them into pyegeria's cached `settings.User_Profile` (screens read the user from there, so every screen opened afterwards runs its Egeria calls as the new user), and exports `EGERIA_USER` / `EGERIA_USER_PASSWORD` so pyegeria clients, commands and subprocesses started from the app use them too. It then sets the main screen's subtitle to `User: `, clears `my_profile_data`, calls `refresh_bindings()` (the feedback-log binding depends on the user) and runs `_load_or_create_profile()`.
+- **`change_user_callback`** handles the menu and connection-error routes. If `ChangeUserScreen` is cancelled while `load_error` is set, it calls `_after_profile_load(False)`, which puts the connection error screen back; otherwise a cancel does nothing.
- **What `_process_profile_data` extracts** from the first `My-User-MD` record:
- Contribution Record, which supplies the Karma Points;
- Projects, Teams, Communities and Roles;
@@ -166,6 +183,7 @@ It returns `200` on success and `401` on failure.
| Technology Types | `handle_technology_types_option()` |
| User Bookmarks | `show_my_bookmarks()` |
| My Network | `show_my_network()` |
+| Change User | `ChangeUserScreen()`, callback `change_user_callback` |
| Leave Feedback | `action_feedback()` |
| Feedback Log | `action_view_feedback_log()` (owner only) |
@@ -591,7 +609,7 @@ Screens tell the app what to do next through their `dismiss(...)` value:
| Code | Meaning |
| :--- | :--- |
| `200` | Done / back |
-| `201` | Alternative navigation |
+| `201` | Alternative navigation (Change User, from the connection error screen) |
| `210` | Quit to main |
| `211` | Subscribe |
| `212` | Sample data |
@@ -616,6 +634,8 @@ The suite in `tests/micro-tests/my_profile/` has one module per handler and scre
**My Network.** `test_network_handler.py` covers `NetworkMixin` and `MyNetworkScreen`, including the remove confirmation; `test_my_egeria.py` checks the menu routing. The feature has also been tested by hand against a live server.
+**Connection errors and Change User.** `test_connection_and_change_user.py` covers `check_egeria_access` (good credentials, unreachable platform, a 401 body returned as the token), `ConnectionErrorScreen` (retry, countdown, buttons), `ChangeUserScreen`, the splash screen's Change User button, and the app's callbacks: reload when Egeria comes back, change user from the error screen and the menu, cancel back to the error screen, and quit with `402`. `test_my_egeria.py` checks that a failed profile load shows the connection error screen instead of exiting, and that `change_user` updates `settings`, the environment and the main screen subtitle.
+
**Add screens.** `test_screens.py::TestBaseAddScreen` covers the `BaseAddScreen` flow: required-field checks, create, link or no link, and keeping the form after a failed create.
---
diff --git a/docs/README.md b/docs/README.md
index 2181dc8c..28516bbe 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -11,7 +11,7 @@ contributors who need more depth than the top-level `README.md`/`CLAUDE.md`/
|---|---|
| `developer_guide_pyegeria_textual.md` | Developer Guide for Python programmers new to Egeria and Textual — step-by-step examples (`examples/developer_guide/`), then My Profile, Dr.Egeria and hey_egeria as case studies. Screenshots in `images/developer-guide/`. |
| `dr_egeria_manual.md` | The Dr.Egeria user manual — command reference, markdown authoring conventions, attribute styles. |
-| `my_egeria_manual.md` | The My Profile App user manual — configuration, starting the app, change user, every screen and key, comments, feedback, return codes, known limitations, running the tests. |
+| `my_egeria_manual.md` | The My Profile App user manual — configuration, starting the app, change user, what happens when Egeria can't be reached, every screen and key, comments, feedback, return codes, known limitations, running the tests. |
| `My-Egeria-Doc.md` | My Profile App reference guide — source layout, mixin structure, routing tables, Egeria calls and report specs per flow, flow and sequence diagrams. |
| `output-formats-and-report-specs.md` | How `generate_output()`/report specs work: `FormatSet`/`Format`/`Column`/`ActionParameter` models, analytic functions, chart output formats. |
| `reference-data-and-valid-metadata-mechanisms.md` | Reference data (business value lists) vs. valid metadata values (allowed values of Egeria's own properties): the types, pyegeria methods, Dr.Egeria Reference Data and Valid Metadata families, and live-verified behaviour. |
diff --git a/docs/developer_guide_pyegeria_textual.md b/docs/developer_guide_pyegeria_textual.md
index d65d9d44..d4009f4e 100644
--- a/docs/developer_guide_pyegeria_textual.md
+++ b/docs/developer_guide_pyegeria_textual.md
@@ -629,6 +629,21 @@ list means the user switched identity, so the background load is cancelled and r
profile exists, the app pushes `CreateProfileScreen`. This is the push → dismiss → callback
pattern from Step 3, used to control the app's whole flow.
+The same pattern handles an Egeria that isn't there. Instead of exiting, a failed load records
+why in `self.load_error`, and the app pushes `ConnectionErrorScreen`. That screen retries on a
+timer (`set_interval`) and dismisses with a code the app's callback acts on: `200` when Egeria
+is back (reload the profile), `201` to change user, `402` to quit:
+
+```python
+ async def connection_error_callback(self, result: int) -> None:
+ if result == 200:
+ await self._load_or_create_profile()
+ elif result == 201:
+ await self.push_screen(ChangeUserScreen(), callback=self.change_user_callback)
+ else:
+ self.exit(402)
+```
+
### 7.4 Loading several tables in parallel with workers
The Shop for Data screen shows five tables, each filled by a separate Egeria query:
diff --git a/docs/my_egeria_manual.md b/docs/my_egeria_manual.md
index 9508cb06..bbd969c9 100644
--- a/docs/my_egeria_manual.md
+++ b/docs/my_egeria_manual.md
@@ -15,17 +15,18 @@ The app's source is in `my_egeria/my_egeria/DemoCode/My_Profile/`.
2. [Configuration](#configuration)
3. [Starting the App](#starting-the-app)
4. [The Splash Screen and Changing User](#the-splash-screen-and-changing-user)
-5. [Creating Your Profile](#creating-your-profile)
-6. [The Main Screen](#the-main-screen)
-7. [Adding and Deleting Elements](#adding-and-deleting-elements)
-8. [Comments and Threaded Responses](#comments-and-threaded-responses)
-9. [Leaving Feedback](#leaving-feedback)
-10. [Other Functions Menu](#other-functions-menu)
-11. [Keyboard Reference](#keyboard-reference)
-12. [Exit and Return Codes](#exit-and-return-codes)
-13. [Troubleshooting](#troubleshooting)
-14. [Known Limitations](#known-limitations)
-15. [Running the Tests](#running-the-tests)
+5. [When Egeria Can't Be Reached](#when-egeria-cant-be-reached)
+6. [Creating Your Profile](#creating-your-profile)
+7. [The Main Screen](#the-main-screen)
+8. [Adding and Deleting Elements](#adding-and-deleting-elements)
+9. [Comments and Threaded Responses](#comments-and-threaded-responses)
+10. [Leaving Feedback](#leaving-feedback)
+11. [Other Functions Menu](#other-functions-menu)
+12. [Keyboard Reference](#keyboard-reference)
+13. [Exit and Return Codes](#exit-and-return-codes)
+14. [Troubleshooting](#troubleshooting)
+15. [Known Limitations](#known-limitations)
+16. [Running the Tests](#running-the-tests)
---
@@ -95,7 +96,33 @@ Behind a reverse proxy, set `MY_PROFILE_PUBLIC_URL` (or `MY_EGERIA_PUBLIC_URL`)
When the app starts it shows a welcome card ("Welcome to My_Profile for Egeria Users, *user*!") while your profile loads from Egeria in the background.
- **Continue to App** (or `Escape`) closes the splash screen. If your profile was found, the main screen is already populated; if not, the app offers to create one (see below).
-- **Change User** replaces the card with *User Name* and *Password* fields and a **Submit** button. Both fields are required. On submit the app discards the profile it was loading and reloads everything for the new user. Every screen opened afterwards uses the new user.
+- **Change User** opens the **Change User** screen (below). If you cancel it you are back on the splash screen.
+
+### The Change User screen
+
+You can switch to a different Egeria user at any time: from the splash screen's **Change User** button, from **Change User** in the main screen's Other Functions menu, or from the [connection error screen](#when-egeria-cant-be-reached).
+
+The screen shows who you are logged in as, with *User Name* and *Password* fields. Press `Enter` in the user name field to move to the password; `Enter` in the password field (or **Submit**) submits. Both fields are required.
+
+Before switching, the app asks Egeria to log in with the new user and password. If Egeria can't be reached or rejects them, a message says why and you stay on the screen to correct them. Once Egeria accepts them, the app discards the current profile and reloads everything for the new user; the main screen's subtitle shows the new user ID. Every screen opened afterwards, and any pyegeria command started from the app, runs as the new user.
+
+**Cancel** (or `Escape`) closes the screen without changing anything.
+
+## When Egeria Can't Be Reached
+
+If the app can't load your profile at startup, because the Egeria platform isn't running or doesn't answer, or because it rejects your user ID and password, it doesn't exit. It shows a **connection error screen** instead, with:
+
+- the platform URL, view server and user it tried;
+- the reason, in red (for example *Unable to connect to the Egeria platform at https://localhost:9443.*, or Egeria's own message when the credentials are rejected);
+- a countdown to the next attempt and the number of attempts so far.
+
+The app tries again every 10 seconds. As soon as Egeria accepts your credentials, the screen closes and your profile loads; if you have no profile yet, the app offers to create one. So you can start the app first and your Egeria instance afterwards.
+
+| Button | Key | Action |
+| :--- | :--- | :--- |
+| **Retry Now** | `r` | Try again straight away |
+| **Change User** | `u` | Open the [Change User screen](#the-change-user-screen). If you cancel it, you return to the connection error screen |
+| **Quit** | `q` | Exit the app (code `402`) |
## Creating Your Profile
@@ -262,6 +289,9 @@ Your network is the people linked to your profile in Egeria. The screen lists th
Each person in your network is linked to your profile with a `Peer` relationship. The nickname and description are stored on that relationship.
+### Change User
+Log in as a different Egeria user and reload the app for them. See [The Change User screen](#the-change-user-screen).
+
### Leave Feedback / Feedback Log
See [Leaving Feedback](#leaving-feedback).
@@ -275,7 +305,8 @@ Every screen follows the same rules:
| Key | Action | Where |
| :--- | :--- | :--- |
-| `Escape` | Back | Every screen except Main; continue on the splash screen; cancel on confirm dialogs |
+| `Escape` | Back | Every screen except Main; continue on the splash screen; cancel on confirm dialogs and Change User |
+| `r` / `u` / `q` | Retry now / Change User / Quit | Connection error screen |
| `q` | Main menu / quit the app | Every list screen / Main screen |
| `Enter` | Select the highlighted row; continue | Tables and trees; Status screen |
| `ctrl+s` | Save / submit the form | Add screens, Add Comment, Create Subscription, Create/Edit Profile, Feedback |
@@ -310,7 +341,7 @@ When the app exits it returns a code; screens also use codes to talk to the app.
| `212` | Sample data request |
| `250` | Show responses to a comment |
| `400` / `401` | Generic failure / profile action failed |
-| `402` | Profile could not be loaded at startup |
+| `402` | Egeria could not be reached (or rejected the credentials) and you chose **Quit** on the connection error screen |
| `403` | Profile creation cancelled or failed |
| `412` / `413` | Error / empty result reloading a newly created profile |
| `416` | Technology types could not be fetched |
@@ -323,14 +354,14 @@ The full list is in `RETURN_CODES.md` in the app folder.
**The app exits immediately with code 430.** Your environment has an old pydantic, often because a globally installed `textual` was used. Activate the project's `.venv` and run `textual run --dev my_egeria.py` or `python my_egeria.py` from there.
-**The app exits with code 402.** The profile could not be retrieved. Check that:
-- `EGERIA_PLATFORM_URL` is correct and reachable (VPN and firewall included);
+**The connection error screen keeps retrying.** The profile could not be retrieved (`402` if you then quit). The red message says why. Check that:
+- `EGERIA_PLATFORM_URL` is correct and reachable (VPN and firewall included) – the screen shows the URL being tried;
- the view server named by `EGERIA_VIEW_SERVER` is running;
-- `EGERIA_USER` and `EGERIA_USER_PASSWORD` are correct, and the user is allowed to use the view server.
+- `EGERIA_USER` and `EGERIA_USER_PASSWORD` are correct, and the user is allowed to use the view server. If the credentials are the problem, press **Change User** rather than restarting the app.
**Something I added doesn't appear.** Press `ctrl+r` on the main screen. Communities you create are not linked to you yet (see Known Limitations), so they don't appear in User Associations.
-**The app quit unexpectedly.** The app only exits by itself in a few cases: a startup problem (codes `402`, `403`, `430`), a failure fetching technology types (`416`), or an error looking up a glossary term (`440`). A failed team lookup is reported as a message instead.
+**The app quit unexpectedly.** The app only exits by itself in a few cases: a startup problem (codes `403`, `430`; `402` only when you press Quit on the connection error screen), a failure fetching technology types (`416`), or an error looking up a glossary term (`440`). A failed team lookup is reported as a message instead.
## Known Limitations
diff --git a/my_egeria/my_egeria/DemoCode/My_Profile/ChangeUserScreen.py b/my_egeria/my_egeria/DemoCode/My_Profile/ChangeUserScreen.py
new file mode 100644
index 00000000..b11868a1
--- /dev/null
+++ b/my_egeria/my_egeria/DemoCode/My_Profile/ChangeUserScreen.py
@@ -0,0 +1,80 @@
+"""
+ PDX-License-Identifier: Apache-2.0
+ Copyright Contributors to the ODPi Egeria project.
+
+ This file provides a screen to change the Egeria user that my_egeria is logged in as.
+
+"""
+
+from typing import Any
+from textual import on
+from textual.app import ComposeResult
+from textual.containers import Horizontal, Vertical
+from textual.screen import ModalScreen
+from textual.widgets import Header, Footer, Button, Label, Input
+
+from pyegeria import load_app_config, settings
+from profile_utils import check_egeria_access
+
+
+class ChangeUserScreen(ModalScreen):
+ """Collect a new user id and password, and check Egeria accepts them.
+
+ Dismisses with [user_name, password] once Egeria has accepted the credentials,
+ or 200 if the user cancels.
+ """
+
+ BINDINGS = [("escape", "cancel", "Cancel")]
+
+ CSS_PATH = "my_profile.tcss"
+
+ def __init__(self, *args: Any, **kwargs: Any) -> None:
+ super().__init__(*args, **kwargs)
+ self.title = "Egeria"
+ self.sub_title = "Change User"
+ load_app_config()
+ self.user_name = settings.User_Profile.user_name
+ self.view_server = settings.Environment.egeria_view_server
+ self.platform_url = settings.Environment.egeria_platform_url
+
+ def compose(self) -> ComposeResult:
+ yield Header(show_clock=True)
+ with Vertical(id="change_user_card"):
+ yield Label(f"Currently logged in as: {self.user_name}\n"
+ f"Enter the user id and password to switch to.", id="change_user_text")
+ yield Input(placeholder="Enter new user name", id="new_user_name")
+ yield Input(placeholder="Enter password for new user", id="new_user_password", password=True)
+ yield Horizontal(
+ Button("Submit", id="submit_new_user", variant="success"),
+ Button("Cancel", id="cancel_change_user", variant="warning"),
+ id="change_user_buttons",
+ )
+ yield Footer()
+
+ def on_mount(self) -> None:
+ self.query_one("#new_user_name", Input).focus()
+
+ def action_cancel(self) -> None:
+ self.dismiss(200)
+
+ @on(Button.Pressed, "#cancel_change_user")
+ def handle_cancel(self) -> None:
+ self.dismiss(200)
+
+ @on(Input.Submitted, "#new_user_name")
+ def handle_user_name_submitted(self) -> None:
+ self.query_one("#new_user_password", Input).focus()
+
+ @on(Input.Submitted, "#new_user_password")
+ @on(Button.Pressed, "#submit_new_user")
+ def handle_submit_new_user(self) -> None:
+ new_user = self.query_one("#new_user_name", Input).value.strip()
+ new_password = self.query_one("#new_user_password", Input).value
+ if not new_user or not new_password:
+ self.notify("Please enter both user name and password", severity="error", timeout=20)
+ return
+ error = check_egeria_access(self.view_server, self.platform_url, new_user, new_password)
+ if error:
+ self.notify(f"User credential validation failed: {error}", severity="error", timeout=30)
+ return
+ self.dismiss([new_user, new_password])
diff --git a/my_egeria/my_egeria/DemoCode/My_Profile/ConnectionErrorScreen.py b/my_egeria/my_egeria/DemoCode/My_Profile/ConnectionErrorScreen.py
new file mode 100644
index 00000000..31a631d8
--- /dev/null
+++ b/my_egeria/my_egeria/DemoCode/My_Profile/ConnectionErrorScreen.py
@@ -0,0 +1,124 @@
+"""
+ PDX-License-Identifier: Apache-2.0
+ Copyright Contributors to the ODPi Egeria project.
+
+ This file provides the screen my_egeria shows while it cannot reach Egeria.
+
+"""
+
+from typing import Any
+from textual import on
+from textual.app import ComposeResult
+from textual.containers import Horizontal, Vertical
+from textual.screen import ModalScreen
+from textual.widgets import Header, Footer, Button, Label, Static
+
+from pyegeria import load_app_config, settings
+from profile_utils import check_egeria_access
+
+# Seconds between automatic reconnection attempts
+RETRY_INTERVAL = 10
+
+
+class ConnectionErrorScreen(ModalScreen):
+ """Explain that Egeria could not be reached, and wait for it to come up.
+
+ The screen retries every RETRY_INTERVAL seconds (or when Retry Now is pressed),
+ and dismisses with 200 as soon as Egeria accepts the current user's credentials.
+ It dismisses with 201 if the user wants to change user, or 402 if they quit.
+ """
+
+ BINDINGS = [
+ ("r", "retry_now", "Retry Now"),
+ ("u", "change_user", "Change User"),
+ ("q", "quit_app", "Quit"),
+ ]
+
+ CSS_PATH = "my_profile.tcss"
+
+ def __init__(self, error_message: str, *args: Any, **kwargs: Any) -> None:
+ super().__init__(*args, **kwargs)
+ self.title = "Egeria"
+ self.sub_title = "Unable to connect to Egeria"
+ self.error_message = error_message
+ self.attempts = 0
+ self.countdown = RETRY_INTERVAL
+ load_app_config()
+ self.user_name = settings.User_Profile.user_name
+ self.user_password = settings.User_Profile.user_pwd
+ self.view_server = settings.Environment.egeria_view_server
+ self.platform_url = settings.Environment.egeria_platform_url
+
+ def compose(self) -> ComposeResult:
+ yield Header(show_clock=True)
+ with Vertical(id="connection_error_card"):
+ yield Label(
+ "my_egeria could not connect to Egeria.\n\n"
+ f"Platform URL: {self.platform_url}\n"
+ f"View Server: {self.view_server}\n"
+ f"User: {self.user_name}\n\n"
+ "Please start your Egeria instance; my_egeria will keep retrying "
+ "and carry on as soon as it is available.",
+ id="connection_error_text",
+ )
+ yield Static(self.error_message, id="connection_error_reason")
+ yield Static(self._status_text(), id="connection_retry_status")
+ yield Horizontal(
+ Button("Retry Now", id="retry_now", variant="success"),
+ Button("Change User", id="change_user", variant="warning"),
+ Button("Quit", id="quit_app", variant="error"),
+ id="connection_error_buttons",
+ )
+ yield Footer()
+
+ def on_mount(self) -> None:
+ self.retry_timer = self.set_interval(1, self._tick)
+
+ def _status_text(self) -> str:
+ attempts = f"Attempts so far: {self.attempts}. " if self.attempts else ""
+ return f"{attempts}Retrying in {self.countdown} seconds..."
+
+ def _tick(self) -> None:
+ self.countdown -= 1
+ if self.countdown <= 0:
+ self.retry_connection()
+ else:
+ self.query_one("#connection_retry_status", Static).update(self._status_text())
+
+ def retry_connection(self) -> None:
+ """Try Egeria again; leave this screen if it now accepts the credentials."""
+ self.attempts += 1
+ self.query_one("#connection_retry_status", Static).update("Trying to connect to Egeria...")
+ error = check_egeria_access(self.view_server, self.platform_url, self.user_name, self.user_password)
+ self.log(f"Connection attempt {self.attempts}: {error or 'connected'}")
+ if error is None:
+ self.retry_timer.stop()
+ self.dismiss(200)
+ return
+ self.error_message = error
+ self.query_one("#connection_error_reason", Static).update(error)
+ self.countdown = RETRY_INTERVAL
+ self.query_one("#connection_retry_status", Static).update(self._status_text())
+
+ def action_retry_now(self) -> None:
+ self.retry_connection()
+
+ def action_change_user(self) -> None:
+ self.retry_timer.stop()
+ self.dismiss(201)
+
+ def action_quit_app(self) -> None:
+ self.retry_timer.stop()
+ self.dismiss(402)
+
+ @on(Button.Pressed, "#retry_now")
+ def handle_retry_now(self) -> None:
+ self.action_retry_now()
+
+ @on(Button.Pressed, "#change_user")
+ def handle_change_user(self) -> None:
+ self.action_change_user()
+
+ @on(Button.Pressed, "#quit_app")
+ def handle_quit(self) -> None:
+ self.action_quit_app()
diff --git a/my_egeria/my_egeria/DemoCode/My_Profile/MainScreen.py b/my_egeria/my_egeria/DemoCode/My_Profile/MainScreen.py
index aff78958..0b59d401 100644
--- a/my_egeria/my_egeria/DemoCode/My_Profile/MainScreen.py
+++ b/my_egeria/my_egeria/DemoCode/My_Profile/MainScreen.py
@@ -71,6 +71,7 @@ def compose(self) -> ComposeResult:
Option("[] Technology Types"),
Option("[] User Bookmarks"),
Option("[] My Network"),
+ Option("[] Change User"),
Option("[] Leave Feedback"),
Option("[] Feedback Log"),
id="other_function_list"
diff --git a/my_egeria/my_egeria/DemoCode/My_Profile/RETURN_CODES.md b/my_egeria/my_egeria/DemoCode/My_Profile/RETURN_CODES.md
index fcdbce60..ee843555 100644
--- a/my_egeria/my_egeria/DemoCode/My_Profile/RETURN_CODES.md
+++ b/my_egeria/my_egeria/DemoCode/My_Profile/RETURN_CODES.md
@@ -26,6 +26,7 @@ In accordance with application conventions:
|---|---|---|---|
| **`400`** | `GENERIC_ERROR` / `BAD_RESULT` | General error code or user cancellation with bad result. Used in StatusScreen when copying GUID fails or user indicates bad status, and in edit elements when invalid operations occur. | `StatusScreen.py`, `EditElementsScreens.py`, `ShowCommentsScreen.py` |
| **`401`** | `PROFILE_ACTION_FAILED` | Profile creation or profile element editing cancelled, rejected, or failed. | `CreateProfileScreen.py`, `EditElementsScreens.py` |
+| **`402`** | `EGERIA_UNAVAILABLE_QUIT` | The profile could not be loaded (Egeria unreachable, or the credentials rejected) and the user chose Quit on the connection error screen. That screen dismisses with 200 once Egeria accepts the current credentials (the app then reloads the profile), or 201 for Change User. | `my_egeria.py`, `ConnectionErrorScreen.py` |
| **`404`** | `TECH_TYPE_NO_PROCESS_FALLBACK_FAIL` | Failed to mount empty processes option list when fallback handler encountered an unexpected error. | `TechnologyTypeScreens.py` |
| **`405`** | `TECH_TYPE_NO_PROCESS_CREATE_FAIL` | Failed to create empty processes option list during initial widget mount. | `TechnologyTypeScreens.py` |
| **`406`** | `TECH_TYPE_PROCESS_CREATE_FAIL` | Error creating technology type governance action processes option list widget. | `TechnologyTypeScreens.py` |
diff --git a/my_egeria/my_egeria/DemoCode/My_Profile/SplashScreen.py b/my_egeria/my_egeria/DemoCode/My_Profile/SplashScreen.py
index 47c2a973..32a2de56 100644
--- a/my_egeria/my_egeria/DemoCode/My_Profile/SplashScreen.py
+++ b/my_egeria/my_egeria/DemoCode/My_Profile/SplashScreen.py
@@ -11,9 +11,10 @@
from textual.app import ComposeResult
from textual.containers import Horizontal, Vertical
from textual.screen import ModalScreen
-from textual.widgets import Header, Footer, Button, Label, Input, Static
+from textual.widgets import Header, Footer, Button, Label, Static
-from pyegeria import Egeria, load_app_config, settings, PyegeriaException
+from pyegeria import load_app_config, settings
+from ChangeUserScreen import ChangeUserScreen
class SplashScreen(ModalScreen):
@@ -71,35 +72,10 @@ def handle_continue(self):
@on(Button.Pressed, "#change_user")
def handle_change_user(self):
- self.query_one("#splash-card").remove_children()
- self.query_one("#change_user").remove()
- self.query_one("#splash-card").mount(
- Input(placeholder="Enter new user name", id="new_user_name"),
- Input(placeholder="Enter password for new user", id="new_user_password", password=True),
- Button("Submit", id="submit_new_user", variant="success")
- )
-
- @on(Button.Pressed, "#submit_new_user")
- def handle_submit_new_user(self):
- self.new_user = self.query_one("#new_user_name", Input).value
- self.new_password = self.query_one("#new_user_password", Input).value
- if not self.new_user or not self.new_password:
- self.notify("Please enter both user name and password", severity="error", timeout=20)
- return
- # Validate new user credentials
- eclient = None
- try:
- eclient = Egeria(
- self.view_server,
- self.platform_url,
- self.new_user,
- self.new_password,
- )
- except PyegeriaException as e:
- self.notify(f"User credential validation failed: {e}", timeout=30, severity="error")
- return
- finally:
- if eclient:
- eclient.close_session()
- self.dismiss([self.new_user, self.new_password])
+ self.app.push_screen(ChangeUserScreen(), callback=self.change_user_callback)
+
+ def change_user_callback(self, result: Any) -> None:
+ """Pass validated [user, password] on to MyProfileApp.mainline; on cancel stay on the splash."""
+ if isinstance(result, list):
+ self.dismiss(result)
diff --git a/my_egeria/my_egeria/DemoCode/My_Profile/my_egeria.py b/my_egeria/my_egeria/DemoCode/My_Profile/my_egeria.py
index 79710c50..e26ed355 100644
--- a/my_egeria/my_egeria/DemoCode/My_Profile/my_egeria.py
+++ b/my_egeria/my_egeria/DemoCode/My_Profile/my_egeria.py
@@ -5,6 +5,7 @@
This file provides the main entry point and core lifecycle for the My Profile Textual App.
"""
+import os
import sys
from pathlib import Path
from typing import Any
@@ -24,6 +25,7 @@
settings,
MyProfile,
PyegeriaException,
+ PyegeriaConnectionException,
print_basic_exception,
exec_report_spec, Egeria,
)
@@ -32,6 +34,8 @@
from textual.binding import Binding
from textual.widgets import DataTable, OptionList, Header, Footer
from SplashScreen import SplashScreen
+from ConnectionErrorScreen import ConnectionErrorScreen
+from ChangeUserScreen import ChangeUserScreen
from CreateProfileScreen import CreateProfileScreen
from EditElementsScreens import (
EditProfileScreen,
@@ -79,6 +83,7 @@
bools_to_strings,
extract_glossary_terms,
check_request_serialization,
+ check_egeria_access,
element_summary,
)
from tech_types_handler import TechTypesMixin
@@ -205,6 +210,7 @@ def __init__(self, *args, **kwargs):
self.user_data = {}
self.user_identities = []
self.user_identity = []
+ self.load_error: str | None = None # why the last profile load could not reach Egeria
def compose(self) -> ComposeResult:
yield Header(show_clock=True)
@@ -232,14 +238,7 @@ async def mainline(self, splash_return: Any = None) -> None:
if isinstance(splash_return, list):
if hasattr(self, "_load_task") and not self._load_task.done():
self._load_task.cancel()
- self.user_name = splash_return[0]
- self.user_password = splash_return[1]
- # Screens read the user from pyegeria's (cached) settings, so record the
- # switch there too, or their Egeria calls would still run as the old user
- settings.User_Profile.user_name = self.user_name
- settings.User_Profile.user_pwd = self.user_password
- self.refresh_bindings()
- await self._load_or_create_profile()
+ await self.change_user(splash_return[0], splash_return[1])
else:
# Splash screen finished (timeout or dismissed)
if hasattr(self, "_load_task"):
@@ -250,22 +249,47 @@ async def mainline(self, splash_return: Any = None) -> None:
except Exception as e:
self.log(f"Error in parallel initial load: {e}")
profile_found = False
+ await self._after_profile_load(profile_found)
+
+ async def change_user(self, user_name: str, user_password: str) -> None:
+ """Switch the app to new (already validated) credentials and reload the profile for that user."""
+ self.log(f"Changing user from {self.user_name} to {user_name}")
+ self.user_name = user_name
+ self.user_password = user_password
+ # Screens read the user from pyegeria's (cached) settings, so record the
+ # switch there too, or their Egeria calls would still run as the old user
+ settings.User_Profile.user_name = user_name
+ settings.User_Profile.user_pwd = user_password
+ # ...and pyegeria clients, commands and subprocesses started from this app read the environment
+ os.environ["EGERIA_USER"] = user_name
+ os.environ["EGERIA_USER_PASSWORD"] = user_password
+ self.get_screen("main").sub_title = f"User: {user_name}"
+ self.my_profile_data = None
+ self.refresh_bindings()
+ await self._load_or_create_profile()
- if not profile_found and (not hasattr(self, "my_profile_data") or not self.my_profile_data):
- self.log("No profile found. Prompting to create one...")
- await self.push_screen(
- CreateProfileScreen(
- user=self.user_name,
- password=self.user_password,
- view_server=self.view_server,
- platform_url=self.platform_url,
- ),
- callback=self.new_profile_return,
- )
+ async def change_user_callback(self, result: Any) -> None:
+ """Return from the Change User screen: a [user, password] list, or 200 if cancelled."""
+ if isinstance(result, list):
+ await self.change_user(result[0], result[1])
+ elif self.load_error:
+ # Change User was chosen from the connection error screen; go back to waiting for Egeria
+ await self._after_profile_load(False)
+
+ async def connection_error_callback(self, result: int) -> None:
+ """Return from the connection error screen: 200 Egeria is back, 201 change user, otherwise quit."""
+ self.log(f"Connection error screen returned: {result}")
+ if result == 200:
+ await self._load_or_create_profile()
+ elif result == 201:
+ await self.push_screen(ChangeUserScreen(), callback=self.change_user_callback)
+ else:
+ self.exit(402)
async def _load_profile_and_populate(self) -> bool:
"""Retrieve the user's profile and populate UI tables in parallel.
Returns True if profile exists and tables were populated, False if profile is missing."""
+ self.load_error = None
try:
self.my_profile_inst = MyProfile(self.view_server, self.platform_url, self.user_name, self.user_password)
self.my_profile_inst.create_egeria_bearer_token(self.user_name, self.user_password)
@@ -278,10 +302,17 @@ async def _load_profile_and_populate(self) -> bool:
if isinstance(self.my_profile_raw, dict) else self.my_profile_raw
)
self.log(f"retrieve profile result: {self.my_profile_data}")
+ except PyegeriaConnectionException as e:
+ self.log(f"Unable to connect to Egeria: {e!s}")
+ self.load_error = f"Unable to connect to the Egeria platform at {self.platform_url}."
+ return False
except PyegeriaException as e:
self.log(f"Error retrieving profile: {e!s}")
print_basic_exception(e)
- self.exit(402)
+ # str(e) is often empty (e.g. on a 401), so ask Egeria directly what is wrong
+ reason = (check_egeria_access(self.view_server, self.platform_url, self.user_name, self.user_password)
+ or str(e) or type(e).__name__)
+ self.load_error = f"Error retrieving the profile for {self.user_name}: {reason}"
return False
if not self.my_profile_data:
@@ -295,6 +326,14 @@ async def _load_profile_and_populate(self) -> bool:
async def _load_or_create_profile(self) -> None:
"""Load user profile; if missing, prompt to create it."""
profile_found = await self._load_profile_and_populate()
+ await self._after_profile_load(profile_found)
+
+ async def _after_profile_load(self, profile_found: bool) -> None:
+ """Wait for Egeria if the load could not reach it; otherwise offer to create a missing profile."""
+ if self.load_error:
+ self.log(f"Profile load failed: {self.load_error}")
+ await self.push_screen(ConnectionErrorScreen(self.load_error), callback=self.connection_error_callback)
+ return
if not profile_found and (not hasattr(self, "my_profile_data") or not self.my_profile_data):
self.log("No profile found. Prompting to create one...")
await self.push_screen(
@@ -681,6 +720,8 @@ async def handle_option_selected(self, event: OptionList.OptionSelected) -> None
self.show_my_bookmarks()
elif selected_option == "My Network":
self.show_my_network()
+ elif selected_option == "Change User":
+ await self.push_screen(ChangeUserScreen(), callback=self.change_user_callback)
elif selected_option == "Subscriptions":
await self.push_screen(ViewSubscriptionsScreen(), callback=self.view_subscriptions_callback)
elif selected_option == "Leave Feedback":
diff --git a/my_egeria/my_egeria/DemoCode/My_Profile/my_profile.tcss b/my_egeria/my_egeria/DemoCode/My_Profile/my_profile.tcss
index e49aeef0..21cf56f2 100644
--- a/my_egeria/my_egeria/DemoCode/My_Profile/my_profile.tcss
+++ b/my_egeria/my_egeria/DemoCode/My_Profile/my_profile.tcss
@@ -284,6 +284,38 @@ Input {
.empty {
background: transparent; /* Invisible structural spacer */
}
+ConnectionErrorScreen, ChangeUserScreen {
+ align: center middle;
+ background: rgba(0, 0, 0, 0.6);
+}
+#connection_error_card, #change_user_card {
+ background: $panel;
+ border: thick $error;
+ padding: 1 2;
+ width: 90;
+ height: auto;
+}
+#change_user_card {
+ border: thick $accent;
+}
+#connection_error_reason {
+ color: $error;
+ margin: 1 0;
+}
+#connection_retry_status {
+ margin-bottom: 1;
+}
+#change_user_card Input {
+ margin-top: 1;
+}
+#connection_error_buttons, #change_user_buttons {
+ height: auto;
+ align: center middle;
+ margin-top: 1;
+}
+#connection_error_buttons Button, #change_user_buttons Button {
+ margin: 0 1;
+}
/* "Link to my profile" row on the Add screens: without these the label takes the
full row width and the Switch is clipped off the right edge */
#add_input_container .link_row {
diff --git a/my_egeria/my_egeria/DemoCode/My_Profile/profile_utils.py b/my_egeria/my_egeria/DemoCode/My_Profile/profile_utils.py
index 9fdfe28c..fa6c0ef7 100644
--- a/my_egeria/my_egeria/DemoCode/My_Profile/profile_utils.py
+++ b/my_egeria/my_egeria/DemoCode/My_Profile/profile_utils.py
@@ -5,6 +5,7 @@
Utility and data transformation functions for My Profile Textual App.
"""
+import json
import re
import sys
from typing import Any
@@ -33,6 +34,39 @@ def check_request_serialization() -> str | None:
"folder (then `textual run my_egeria.py --dev`), or `uv run python my_egeria.py`.")
+def check_egeria_access(view_server: str, platform_url: str, user_name: str, user_password: str) -> str | None:
+ """Return None if Egeria is reachable and accepts the credentials, otherwise a message saying why not.
+
+ A successful create_egeria_bearer_token() is not proof the credentials are good:
+ on a 401 it hands back Egeria's JSON error body as if it were the token, so that
+ case is detected here instead.
+ """
+ from pyegeria import Egeria, PyegeriaConnectionException, PyegeriaException
+
+ eclient = None
+ try:
+ # The client constructor checks that the platform answers
+ eclient = Egeria(view_server, platform_url, user_name, user_password)
+ token = eclient.create_egeria_bearer_token(user_name, user_password)
+ except PyegeriaConnectionException:
+ return f"Unable to connect to the Egeria platform at {platform_url}."
+ except PyegeriaException as e:
+ return f"Egeria at {platform_url} returned an error: {e!s}"
+ finally:
+ if eclient:
+ eclient.close_session()
+
+ if not token or token == "FAILED":
+ return f"Unable to connect to the Egeria platform at {platform_url}."
+ if token.lstrip().startswith("{"):
+ try:
+ reason = json.loads(token).get("exceptionErrorMessage")
+ except ValueError:
+ reason = None
+ return reason or f"Egeria rejected the credentials for user {user_name}."
+ return None
+
+
def truncate_at_sequence(data: Any, target: str = "specificationMermaidGraph") -> tuple[Any, bool]:
"""
Recursively traverses data. Truncates strings containing the target
diff --git a/tests/micro-tests/my_profile/test_connection_and_change_user.py b/tests/micro-tests/my_profile/test_connection_and_change_user.py
new file mode 100644
index 00000000..238f27ca
--- /dev/null
+++ b/tests/micro-tests/my_profile/test_connection_and_change_user.py
@@ -0,0 +1,240 @@
+"""
+ PDX-License-Identifier: Apache-2.0
+ Copyright Contributors to the ODPi Egeria project.
+
+ Tests for waiting on an unreachable Egeria at startup, and for changing the logged in user.
+"""
+
+from unittest.mock import MagicMock, AsyncMock, patch
+import pytest
+from textual.app import App
+from textual.widgets import Button, Input, Static
+
+from my_egeria.DemoCode.My_Profile.my_egeria import MyProfileApp
+from ChangeUserScreen import ChangeUserScreen
+from ConnectionErrorScreen import ConnectionErrorScreen, RETRY_INTERVAL
+from SplashScreen import SplashScreen
+from profile_utils import check_egeria_access
+from pyegeria import PyegeriaConnectionException
+
+BAD_CREDENTIALS_BODY = (
+ '{"class":"FFDCResponseBase","relatedHTTPCode":401,'
+ '"exceptionErrorMessage":"User authentication failed for user: garygeeke - Bad credentials"}'
+)
+
+
+class HostApp(App):
+ """Pushes one screen and records what it dismisses with."""
+
+ def __init__(self, screen_factory):
+ super().__init__()
+ self.screen_factory = screen_factory
+ self.target_screen = None
+ self.dismissed_result = "not dismissed"
+
+ async def on_mount(self):
+ self.target_screen = self.screen_factory()
+
+ def _cb(result):
+ self.dismissed_result = result
+
+ await self.push_screen(self.target_screen, callback=_cb)
+
+
+class TestCheckEgeriaAccess:
+ """check_egeria_access must tell 'unreachable' and 'bad credentials' apart from success."""
+
+ @patch("pyegeria.Egeria")
+ def test_good_credentials(self, mock_egeria):
+ mock_egeria.return_value.create_egeria_bearer_token.return_value = "eyJhbGciOi.jwt.token"
+ assert check_egeria_access("qs-view-server", "https://localhost:9443", "garygeeke", "secret") is None
+ mock_egeria.return_value.close_session.assert_called_once()
+
+ @patch("pyegeria.Egeria")
+ def test_unreachable_platform(self, mock_egeria):
+ mock_egeria.side_effect = PyegeriaConnectionException(None, None, ConnectionError("refused"))
+ error = check_egeria_access("qs-view-server", "https://localhost:9443", "garygeeke", "secret")
+ assert error == "Unable to connect to the Egeria platform at https://localhost:9443."
+
+ @patch("pyegeria.Egeria")
+ def test_bad_credentials_body_is_not_a_token(self, mock_egeria):
+ mock_egeria.return_value.create_egeria_bearer_token.return_value = BAD_CREDENTIALS_BODY
+ error = check_egeria_access("qs-view-server", "https://localhost:9443", "garygeeke", "wrong")
+ assert "Bad credentials" in error
+
+
+class TestConnectionErrorScreen:
+
+ @pytest.mark.asyncio
+ @patch("ConnectionErrorScreen.check_egeria_access", return_value=None)
+ async def test_retry_succeeds_dismisses_200(self, mock_check):
+ app = HostApp(lambda: ConnectionErrorScreen("Unable to connect"))
+ async with app.run_test() as pilot:
+ app.target_screen.action_retry_now()
+ await pilot.pause()
+ assert app.dismissed_result == 200
+
+ @pytest.mark.asyncio
+ @patch("ConnectionErrorScreen.check_egeria_access", return_value="Still down")
+ async def test_retry_fails_stays_and_shows_reason(self, mock_check):
+ app = HostApp(lambda: ConnectionErrorScreen("Unable to connect"))
+ async with app.run_test() as pilot:
+ screen = app.target_screen
+ screen.action_retry_now()
+ await pilot.pause()
+ assert app.dismissed_result == "not dismissed"
+ assert screen.attempts == 1
+ assert screen.countdown == RETRY_INTERVAL
+ assert "Still down" in str(screen.query_one("#connection_error_reason", Static).render())
+
+ @pytest.mark.asyncio
+ @patch("ConnectionErrorScreen.check_egeria_access", return_value=None)
+ async def test_countdown_triggers_retry(self, mock_check):
+ app = HostApp(lambda: ConnectionErrorScreen("Unable to connect"))
+ async with app.run_test() as pilot:
+ app.target_screen.countdown = 1
+ app.target_screen._tick()
+ await pilot.pause()
+ mock_check.assert_called_once()
+ assert app.dismissed_result == 200
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize("button_id,expected", [("change_user", 201), ("quit_app", 402)])
+ @patch("ConnectionErrorScreen.check_egeria_access", return_value="down")
+ async def test_buttons(self, mock_check, button_id, expected):
+ app = HostApp(lambda: ConnectionErrorScreen("Unable to connect"))
+ async with app.run_test(size=(120, 40)) as pilot:
+ app.target_screen.query_one(f"#{button_id}", Button).press()
+ await pilot.pause()
+ assert app.dismissed_result == expected
+
+
+class TestChangeUserScreen:
+
+ async def _submit(self, pilot, app, user, password):
+ app.target_screen.query_one("#new_user_name", Input).value = user
+ app.target_screen.query_one("#new_user_password", Input).value = password
+ app.target_screen.handle_submit_new_user()
+ await pilot.pause()
+
+ @pytest.mark.asyncio
+ @patch("ChangeUserScreen.check_egeria_access", return_value=None)
+ async def test_valid_credentials_dismiss_with_user_and_password(self, mock_check):
+ app = HostApp(ChangeUserScreen)
+ async with app.run_test() as pilot:
+ await self._submit(pilot, app, " erinoverview ", "secret")
+ assert app.dismissed_result == ["erinoverview", "secret"]
+ assert mock_check.call_args[0][2:] == ("erinoverview", "secret")
+
+ @pytest.mark.asyncio
+ @patch("ChangeUserScreen.check_egeria_access", return_value="Bad credentials")
+ async def test_rejected_credentials_stay_on_screen(self, mock_check):
+ app = HostApp(ChangeUserScreen)
+ async with app.run_test() as pilot:
+ await self._submit(pilot, app, "erinoverview", "wrong")
+ assert app.dismissed_result == "not dismissed"
+
+ @pytest.mark.asyncio
+ @patch("ChangeUserScreen.check_egeria_access")
+ async def test_missing_fields_not_checked(self, mock_check):
+ app = HostApp(ChangeUserScreen)
+ async with app.run_test() as pilot:
+ await self._submit(pilot, app, "erinoverview", "")
+ mock_check.assert_not_called()
+ assert app.dismissed_result == "not dismissed"
+
+ @pytest.mark.asyncio
+ async def test_cancel_dismisses_200(self):
+ app = HostApp(ChangeUserScreen)
+ async with app.run_test() as pilot:
+ await pilot.press("escape")
+ assert app.dismissed_result == 200
+
+
+class TestSplashChangeUser:
+
+ @pytest.mark.asyncio
+ async def test_change_user_button_opens_change_user_screen(self):
+ app = HostApp(SplashScreen)
+ async with app.run_test(size=(120, 40)) as pilot:
+ app.target_screen.query_one("#change_user", Button).press()
+ await pilot.pause()
+ assert isinstance(app.screen, ChangeUserScreen)
+
+ @pytest.mark.asyncio
+ async def test_new_credentials_passed_on_cancel_stays(self):
+ app = HostApp(SplashScreen)
+ async with app.run_test() as pilot:
+ app.target_screen.change_user_callback(200)
+ await pilot.pause()
+ assert app.dismissed_result == "not dismissed"
+ app.target_screen.change_user_callback(["erinoverview", "secret"])
+ await pilot.pause()
+ assert app.dismissed_result == ["erinoverview", "secret"]
+
+
+class TestAppConnectionAndUserChange:
+
+ @pytest.mark.asyncio
+ async def test_connection_failure_message_names_platform(self):
+ app = MyProfileApp()
+ with patch("my_egeria.DemoCode.My_Profile.my_egeria.MyProfile",
+ side_effect=PyegeriaConnectionException(None, None, ConnectionError("refused"))):
+ assert await app._load_profile_and_populate() is False
+ assert app.load_error == f"Unable to connect to the Egeria platform at {app.platform_url}."
+
+ @pytest.mark.asyncio
+ async def test_egeria_back_reloads_profile(self):
+ app = MyProfileApp()
+ app._load_or_create_profile = AsyncMock()
+ await app.connection_error_callback(200)
+ app._load_or_create_profile.assert_awaited_once()
+
+ @pytest.mark.asyncio
+ async def test_change_user_from_error_screen(self):
+ app = MyProfileApp()
+ app.push_screen = AsyncMock()
+ await app.connection_error_callback(201)
+ assert isinstance(app.push_screen.call_args[0][0], ChangeUserScreen)
+ assert app.push_screen.call_args.kwargs["callback"] == app.change_user_callback
+
+ @pytest.mark.asyncio
+ async def test_quit_from_error_screen_exits_402(self):
+ app = MyProfileApp()
+ app.exit = MagicMock()
+ await app.connection_error_callback(402)
+ app.exit.assert_called_once_with(402)
+
+ @pytest.mark.asyncio
+ async def test_cancel_change_user_while_disconnected_returns_to_error_screen(self):
+ app = MyProfileApp()
+ app.load_error = "Unable to connect"
+ app.push_screen = AsyncMock()
+ await app.change_user_callback(200)
+ assert isinstance(app.push_screen.call_args[0][0], ConnectionErrorScreen)
+
+ @pytest.mark.asyncio
+ async def test_cancel_change_user_while_connected_does_nothing(self):
+ app = MyProfileApp()
+ app.push_screen = AsyncMock()
+ app.change_user = AsyncMock()
+ await app.change_user_callback(200)
+ app.push_screen.assert_not_called()
+ app.change_user.assert_not_called()
+
+ @pytest.mark.asyncio
+ async def test_change_user_callback_switches_user(self):
+ app = MyProfileApp()
+ app.change_user = AsyncMock()
+ await app.change_user_callback(["erinoverview", "secret"])
+ app.change_user.assert_awaited_once_with("erinoverview", "secret")
+
+ @pytest.mark.asyncio
+ async def test_change_user_menu_option(self):
+ app = MyProfileApp()
+ app.push_screen = AsyncMock()
+ event = MagicMock()
+ event.option.prompt = "[] Change User"
+ await app.handle_option_selected(event)
+ assert isinstance(app.push_screen.call_args[0][0], ChangeUserScreen)
+ assert app.push_screen.call_args.kwargs["callback"] == app.change_user_callback
diff --git a/tests/micro-tests/my_profile/test_my_egeria.py b/tests/micro-tests/my_profile/test_my_egeria.py
index 906c085b..effde34f 100644
--- a/tests/micro-tests/my_profile/test_my_egeria.py
+++ b/tests/micro-tests/my_profile/test_my_egeria.py
@@ -8,6 +8,7 @@
Egeria view server when PYEG_LIVE_EGERIA=1 is set.
"""
+import os
from unittest.mock import MagicMock, AsyncMock, patch, PropertyMock
import pytest
from textual.widgets import Button, OptionList, DataTable
@@ -17,6 +18,7 @@
from MainScreen import MainScreen
from CreateProfileScreen import CreateProfileScreen
from SplashScreen import SplashScreen
+from ConnectionErrorScreen import ConnectionErrorScreen
from UserIdentitiesScreen import UserIdentitiesScreen
from EditElementsScreens import EditProfileScreen
from egeria_backend import at_least, is_int, nonempty_str
@@ -174,17 +176,23 @@ async def test_splash_escape_still_prompts_create_profile(self, backend):
assert isinstance(app.screen, CreateProfileScreen)
@pytest.mark.asyncio
- async def test_app_load_profile_exception_exits_402(self, backend):
+ async def test_app_load_profile_exception_shows_connection_error(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"))
backend.always_fake("my_egeria.DemoCode.My_Profile.my_egeria.MyProfile", returns=mock_mp)
+ backend.always_fake("my_egeria.DemoCode.My_Profile.my_egeria.check_egeria_access", returns="Bad credentials")
app = MyProfileApp()
app.exit = MagicMock()
+ app.push_screen = AsyncMock()
await app._load_or_create_profile()
- app.exit.assert_called_once_with(402)
+ app.exit.assert_not_called()
+ assert app.load_error == f"Error retrieving the profile for {app.user_name}: Bad credentials"
+ screen = app.push_screen.call_args[0][0]
+ assert isinstance(screen, ConnectionErrorScreen)
+ assert app.push_screen.call_args.kwargs["callback"] == app.connection_error_callback
def test_new_profile_return_error_code(self):
app = MyProfileApp()
@@ -275,10 +283,13 @@ def test_status_callback(self):
app.show_main_screen.assert_called_once()
@pytest.mark.asyncio
- async def test_change_user_updates_settings_for_screens(self):
+ async def test_change_user_updates_settings_for_screens(self, monkeypatch):
"""After Change User, screens that read pyegeria settings must see the new user."""
from pyegeria import settings
from AddToElementsScreens import AddTodoScreen
+ # change_user exports the new credentials; monkeypatch puts the environment back afterwards
+ monkeypatch.setenv("EGERIA_USER", "unchanged")
+ monkeypatch.setenv("EGERIA_USER_PASSWORD", "unchanged")
saved = (settings.User_Profile.user_name, settings.User_Profile.user_pwd)
app = MyProfileApp()
app._load_or_create_profile = AsyncMock()
@@ -288,6 +299,8 @@ async def test_change_user_updates_settings_for_screens(self):
assert (app.user_name, app.user_password) == ("erinoverview", "erin-pwd")
screen = AddTodoScreen("todos_table", "user-guid-123")
assert (screen.user_name, screen.user_password) == ("erinoverview", "erin-pwd")
+ assert (os.environ["EGERIA_USER"], os.environ["EGERIA_USER_PASSWORD"]) == ("erinoverview", "erin-pwd")
+ assert app.get_screen("main").sub_title == "User: erinoverview"
app._load_or_create_profile.assert_awaited_once()
finally:
settings.User_Profile.user_name, settings.User_Profile.user_pwd = saved