diff --git a/docs/My-Egeria-Doc.md b/docs/My-Egeria-Doc.md index f90359bd..b51df45a 100644 --- a/docs/My-Egeria-Doc.md +++ b/docs/My-Egeria-Doc.md @@ -28,6 +28,7 @@ All source files named here are in `my_egeria/my_egeria/DemoCode/My_Profile/`. | `tech_types_handler.py`, `TechnologyTypeScreens.py` | `TechTypesMixin` and the technology type screens | | `team_roles_handler.py`, `MyTeamScreen.py` | `TeamRolesMixin` and the team roster screen | | `bookmarks_handler.py`, `MyBookMarksScreen.py` | `BookmarksMixin` (bookmarks kept in a personal collection) and the bookmarks screen | +| `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 | @@ -64,6 +65,13 @@ classDiagram list_my_bookmarks() add_my_bookmark() / delete_my_bookmark() } + class NetworkMixin { + show_my_network() + list_my_peers() + search_people() + link_peer() / unlink_peer() + get_peer_details() + } App <|-- MyProfileApp TechTypesMixin <|-- MyProfileApp ShopForDataMixin <|-- MyProfileApp @@ -71,6 +79,7 @@ classDiagram ElementsCrudMixin <|-- MyProfileApp FeedbackMixin <|-- MyProfileApp BookmarksMixin <|-- MyProfileApp + NetworkMixin <|-- MyProfileApp ``` - **App bindings:** @@ -156,6 +165,7 @@ It returns `200` on success and `401` on failure. | Subscriptions | `ViewSubscriptionsScreen()` | | Technology Types | `handle_technology_types_option()` | | User Bookmarks | `show_my_bookmarks()` | +| My Network | `show_my_network()` | | Leave Feedback | `action_feedback()` | | Feedback Log | `action_view_feedback_log()` (owner only) | @@ -481,6 +491,61 @@ After each change it re-reads the list through the app. The lookup matches on th --- +## My Network + +A user's network is the set of people linked to their profile by a `Peer` relationship. `NetworkMixin` (`network_handler.py`) makes the Egeria calls; `MyNetworkScreen` only drives it through `self.app`. Each call builds a fresh `Egeria` client with a bearer token (`_network_client()`) and closes its session afterwards. + +```mermaid +sequenceDiagram + autonumber + actor User + participant App as NetworkMixin + participant MNS as MyNetworkScreen + participant Confirm as ConfirmRemovePeerScreen + participant EG as Egeria + + User->>App: Other Functions → My Network + App->>EG: get_related_elements(user_GUID, "Peer") + App->>MNS: push(peer rows) + opt view details (v) + MNS->>App: get_peer_details(peer GUID) + App->>EG: get_actor_profile_by_guid(JSON) + App->>App: _generate_my_profile_output(DICT, "My-User") + App-->>MNS: (contact methods, communities) + end + opt add (search, select, Add to Network) + MNS->>App: search_people(name) + App->>EG: find_actor_profiles(DICT, "User-Identities", page_size=10) + App-->>MNS: (name, GUID, qualified name) rows + MNS->>App: link_peer(GUID, nickname, description) + App->>EG: link_peer_person(user_GUID, GUID, PeerProperties) + MNS->>App: list_my_peers() + end + opt remove (d) + MNS->>Confirm: push(peer name) + Confirm-->>MNS: True / False + MNS->>App: unlink_peer(GUID) + App->>EG: unlink_peer_person(user_GUID, GUID) + MNS->>App: list_my_peers() + end +``` + +| Function | What it does | +| :--- | :--- | +| `show_my_network()` | Refuses (with a notice) if `user_GUID` isn't set yet, i.e. the profile hasn't loaded; otherwise pushes `MyNetworkScreen(list_my_peers())` | +| `list_my_peers()` | `get_related_elements(user_GUID, "Peer")`, keeping only elements whose properties are `PersonProperties`. Returns `(qualified name, display name, full name, pronouns, GUID, description)` rows; the GUID comes from the peer's `elementHeader` and the description from `relationshipProperties`. `[]` when Egeria answers "No elements found", `None` if Egeria can't be reached | +| `search_people(search_string)` | `find_actor_profiles(..., ignore_case=True, report_spec="User-Identities", page_size=10)` → `(display name, GUID, qualified name)` rows, excluding the current user. The screen also drops anyone already in the network | +| `link_peer(guid, nickname, description)` | `link_peer_person(user_GUID, guid, NewRelationshipRequestBody)` with `PeerProperties` (`peer_relationship_name` = nickname, `description`) | +| `unlink_peer(guid)` | `unlink_peer_person(user_GUID, guid, DeleteRelationshipRequestBody)`, using the same GUID order as `link_peer_person` | +| `get_peer_details(guid)` | `get_actor_profile_by_guid(guid, JSON)`, then `_generate_my_profile_output(..., report_spec="My-User")`. Returns the `Contact Methods` and `Communities` lists, or `None` with a notice if there are no details | + +- **Removing only unlinks.** The peer's profile is never changed. If the details panel is showing the person just removed, it is hidden. +- **Refresh.** After a successful add or remove the screen re-reads the network with `list_my_peers()`. +- **Errors.** Every Egeria failure is reported with `notify(...)` and the screen stays open; none of them exit the app. +- **Return code.** `MyNetworkScreen` dismisses with `200` on `q`; `my_network_callback` just logs it. + +--- + ## Status Screen `StatusScreen(message)` shows a read-only message: @@ -497,7 +562,8 @@ The app's `status_callback` then returns to the main screen. | Report spec | Used by | | :--- | :--- | | `My-User-MD` | Profile load and reload | -| `User-Identities` | User identity table | +| `User-Identities` | User identity table; My Network people search | +| `My-User` | My Network peer details (contact methods, communities) | | `My-User-ToDos` | To-dos table | | `Actor-Profiles` | User GUID fallback | | `Comment-by-Element` | Comments screen | @@ -509,7 +575,7 @@ The app's `status_callback` then returns to the main screen. | Client | Used for | | :--- | :--- | | `MyProfile` | Profile, identities, to-dos, `add_my_profile` | -| `Egeria` | Element create, link and delete; comments from main; bookmarks; feedback note logs; tabular data; `update_actor_profile`; `find_collections` | +| `Egeria` | Element create, link and delete; comments from main; bookmarks; My Network; feedback note logs; tabular data; `update_actor_profile`; `find_collections` | | `EgeriaTech` | Comments screen | | `AutomatedCuration` | Technology types | | `ProductManager` | Digital subscriptions | @@ -548,6 +614,8 @@ The suite in `tests/micro-tests/my_profile/` has one module per handler and scre **Running the whole app in a test.** Tests that run `MyProfileApp.run_test()` stub the profile client with `stub_profile_client()`. +**My Network.** `test_network_handler.py` covers `NetworkMixin` and `MyNetworkScreen`, including the remove confirmation; `test_my_profile_app.py` checks the menu routing. The feature has also been tested by hand against a live server. + **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/my_profile_app_manual.md b/docs/my_profile_app_manual.md index 0bd49f6d..1065387c 100644 --- a/docs/my_profile_app_manual.md +++ b/docs/my_profile_app_manual.md @@ -3,7 +3,7 @@ # My Profile App User Manual -The **My Profile App** is a terminal user interface (TUI), built with [Textual](https://textual.textualize.io/) and the `pyegeria` SDK, that gives an Egeria user a personal dashboard: their profile, roles, teams, communities, blogs, journal, to-dos and user identities, plus access to comments, feedback, data shopping, subscriptions and technology types. +The **My Profile App** is a terminal user interface (TUI), built with [Textual](https://textual.textualize.io/) and the `pyegeria` SDK, that gives an Egeria user a personal dashboard: their profile, roles, teams, communities, blogs, journal, to-dos and user identities, plus access to comments, feedback, data shopping, subscriptions, technology types and their network of peers. This manual describes how to *use* the app. For how it works internally (screens, handlers, Egeria calls and flow diagrams) see the [My Profile Reference Guide](My-Egeria-Doc.md). @@ -218,7 +218,7 @@ Opens a screen with five tables, each loading in the background: **Glossaries**, - `u` or `ctrl+s` – subscribe to the highlighted item. - `t` – search glossary terms. Back from the search (`g`) returns to Shop for Data. - `k` – bookmark the highlighted item; `ctrl+b` – open bookmarks with its GUID ready to add. -- `b` – back; `q` – quit the screen. +- `b` or `Escape` – back to the main screen; `q` – quit the screen. On the overview screen, `s` subscribes to the selected node, `b` goes back and `q` returns to the main screen. On a sample-data view, `s` subscribes, and `b`, `Escape` or `Enter` go back. @@ -250,6 +250,16 @@ On the bookmarks screen: Your bookmarks are kept in Egeria as the members of your private **My Bookmarks** collection (qualified name `Bookmarks::`), and only that collection. It is created the first time you bookmark something, anchored to your profile and linked from it with a `ResourceList` relationship whose resource use is "Bookmarks". +### My Network +Your network is the people linked to your profile in Egeria. The screen lists them (qualified name, display name, full name, pronouns, GUID and the description you gave the link). If your profile has not finished loading, the app tells you to wait and does not open the screen. + +- **View details:** highlight a person and press `v` (or **Person Details**). A panel shows their **contact methods** and the **communities** they belong to. A message tells you if either is empty. +- **Add someone:** type all or part of a name in the search field and press `Enter`. Up to 10 matching people are listed, leaving out you and anyone already in your network. Select a person, optionally enter a **nickname** and a **description** of your relationship, then press **Add to Network**. **Cancel** clears the selection. +- **Remove someone:** highlight a person and press `d` (or **Remove from Network**), then confirm with **Remove**. **Cancel** or `Escape` keeps them. Only the link between you is removed; their profile is not changed. +- `q` returns to the main screen. + +Each person in your network is linked to your profile with a `Peer` relationship. The nickname and description are stored on that relationship. + ### Leave Feedback / Feedback Log See [Leaving Feedback](#leaving-feedback). @@ -267,10 +277,12 @@ See [Leaving Feedback](#leaving-feedback). | `ctrl+b` | Main screen, Shop for Data | Bookmarks, with the highlighted row's GUID ready to add | | `ctrl+k` | Main screen | Bookmark the selected row | | `ctrl+n` / `d` / `q` | Bookmarks | Add by GUID / remove highlighted / close | +| `v` / `d` / `q` | My Network | Show highlighted person's details / remove highlighted person / close | +| `Enter` | My Network search field | Search for people to add | | `a` / `d` / `Escape` | Edit screens | Add row / delete row / back | | `ctrl+a` / `q` | Add screens | Add / close (outside a text field) | | `ctrl+a` / `ctrl+r` / `q` | Comments screen | Add comment / show responses to highlighted comment / close | -| `s` / `u` or `ctrl+s` / `t` / `k` / `b` / `q` | Shop for Data | Sample / subscribe / search glossary terms / bookmark / back / close | +| `s` / `u` or `ctrl+s` / `t` / `k` / `b` or `Escape` / `q` | Shop for Data | Sample / subscribe / search glossary terms / bookmark / back / close | | `s` / `b` / `q` | Overview | Subscribe / back / back to main | | `c` / `q` | Create Subscription | Create / cancel | | `c` / `Enter` / `q` | Status screen | Copy GUID / close / close | diff --git a/my_egeria/my_egeria/DemoCode/Experimental Code/my_network.py b/my_egeria/my_egeria/DemoCode/Experimental Code/my_network.py new file mode 100644 index 00000000..2fe435aa --- /dev/null +++ b/my_egeria/my_egeria/DemoCode/Experimental Code/my_network.py @@ -0,0 +1,584 @@ +""" + PDX-License-Identifier: Apache-2.0 + Copyright Contributors to the ODPi Egeria project. + + This file provides the main entry point and core lifecycle for the My Profile Textual App. +""" + +import sys +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from textual import on +from textual.app import App, ComposeResult +from textual.binding import Binding +from textual.containers import ScrollableContainer, Grid +from textual.css.query import NoMatches +from textual.screen import ModalScreen +from textual.widgets import Header, Footer, Static, Placeholder, Input, DataTable, Button, Label + +# Add the project root to sys.path to allow running this script from any directory +root_path = Path(__file__).resolve().parents[4] +if str(root_path) not in sys.path: + sys.path.append(str(root_path)) + +current_dir = Path(__file__).resolve().parent +if str(current_dir) not in sys.path: + sys.path.append(str(current_dir)) + +import os +os.environ.setdefault("PYEGERIA_CONFIG_DIRECTORY", str(root_path / "config")) + +from pyegeria import ( + load_app_config, + settings, + MyProfile, + PyegeriaException, + print_basic_exception, + exec_report_spec, Egeria, +) + +class ConfirmRemoveScreen(ModalScreen[bool]): + """Modal dialog asking the user to confirm removing a peer from their network. + Dismisses with True if the user confirms, False otherwise.""" + + DEFAULT_CSS = """ + ConfirmRemoveScreen { + align: center middle; + } + #confirm_dialog { + grid-size: 2; + grid-gutter: 1 2; + grid-rows: 1fr 3; + padding: 0 1; + width: 60; + height: 11; + border: thick $background 80%; + background: $surface; + } + #confirm_question { + column-span: 2; + height: 1fr; + width: 1fr; + content-align: center middle; + } + #confirm_dialog Button { + width: 100%; + } + """ + + BINDINGS = [("escape", "cancel", "Cancel")] + + def __init__(self, question: str) -> None: + super().__init__() + self.question = question + + def compose(self) -> ComposeResult: + yield Grid( + Label(self.question, id="confirm_question"), + Button("Remove", variant="error", id="confirm_remove"), + Button("Cancel", variant="primary", id="confirm_cancel"), + id="confirm_dialog", + ) + + @on(Button.Pressed, "#confirm_remove") + def handle_confirm(self, event: Button.Pressed) -> None: + self.dismiss(True) + + @on(Button.Pressed, "#confirm_cancel") + def handle_cancel(self, event: Button.Pressed) -> None: + self.dismiss(False) + + def action_cancel(self) -> None: + self.dismiss(False) + + +class MyNetworkApp(App): + """My Network App. + + Finds user ids based on a search parameter to allow the current user to add the user(s) + found to their private "peer network" in their profile in Egeria. + """ + + BINDINGS = [ + ("q", "quit", "Quit"), + ("r", "refresh", "Refresh Data"), + Binding("ctrl+f", "feedback", "Feedback", priority=True), + Binding("f3", "view_feedback_log", "Feedback Log", priority=True), + ] + + # CSS_PATH = "my_profile.tcss" + + def __init__(self, *args: Any, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self.title = "My_Network" + self.sub_title = "Egeria network of contacts for the current user" + self.description = "Display the user's network of (Egeria) contacts." + + app_config = settings.Environment + self.log(f"Application Config: {app_config}") + app_user = settings.User_Profile + self.log(f"User Profile: {app_user}") + self.user_name = app_user.user_name or "tessatube" + self.user_password = app_user.user_pwd or "secret" + self.view_server = app_config.egeria_view_server or "qs-view-server" + self.platform_url = app_config.egeria_platform_url or "https://127.0.0.1:9443" + self.log(f"Platform URL: {self.platform_url}") + self.log(f"View Server: {self.view_server}") + self.log(f"User: {self.user_name}") + self.log(f"User PWD: {self.user_password}") + self.users_found: DataTable = DataTable(id="users_found") + self.my_network: DataTable = DataTable(id="my_network") + self.my_user_contact_details: DataTable = DataTable(id="my_user_contact_details") + self.my_user_communities: DataTable = DataTable(id="my_user_communities") + self.nickname = "" + self.description = "" + self.selected_network_member = None + self.peer_qname = None, + self.peer_dname = None, + self.peer_fname = None, + self.peer_pronouns = None, + self.peer_guid = None, + self.peer_description = None + # GUID of the peer whose details are currently displayed (get_peer_data resets self.peer_guid) + self.details_peer_guid = None + + def on_mount(self) -> None: + """Perform actions when the application is mounted.""" + self.my_network.add_columns("Qualified Name", "Display Name", "Full Name", "Pronouns", "GUID", "Description") + self.my_network.cursor_type="row" + self.my_network.zebra_stripes=True + error_flag = 0 + try: + eclient = Egeria(self.view_server, self.platform_url, self.user_name, self.user_password) + token=eclient.create_egeria_bearer_token(self.user_name, self.user_password) + self.my_profile=eclient.get_my_profile(output_format="DICT") + self.log(f"User profile retrieved successfully, {self.my_profile}") + except PyegeriaException as e: + self.log(f"Error: {e}") + self.notify(f"Error, unable to retrieve user profile: {e}", severity="error", timeout=20) + error_flag = 1 + finally: + if eclient: + eclient.close_session() + if error_flag == 1: + # Use the error_flag to see if there was an error and if so exit + self.exit(400) + if isinstance(self.my_profile, list): + self.user_GUID = self.my_profile[0].get("GUID") or None + else: + self.notify(f"User profile is not a list: {self.my_profile}", severity="error", timeout=20) + self.exit(400) + self.get_peer_data() + return + + + def get_peer_data(self): + # clear out DataTable + self.my_network.clear() + # Get existing peer relationship list + try: + eclient = Egeria(self.view_server, self.platform_url, self.user_name, self.user_password) + token = eclient.create_egeria_bearer_token(self.user_name, self.user_password) + open_metadata_type_name = "Referenceable" + relationship_type = "Peer" + body = { + "class": "ResultsRequestBody", + "metadataElementTypeName": open_metadata_type_name, + "includeOnlyRelationships": ["ContactThrough"] + } + peers = eclient.get_related_elements(element_guid=self.user_GUID, + relationship_type=relationship_type, + body=body) + self.log(f"Peers: {peers}") + except PyegeriaException as e: + self.log(f"Error creating peer relationship: {e}") + self.notify(f"Error creating peer relationship: {e}", timeout=15, severity="error") + self.exit(400) + finally: + if eclient: + eclient.close_session() + if peers != "No elements found": + for peer in peers: + self.log(f"Peer: {peer}") + peer_properties = peer.get("properties", None) + self.log(f"Peer Properties: {peer_properties}") + peer_related_by = peer.get("relatedBy", None) + self.log(f"Peer Related By: {peer_related_by}") + peer_relationship_properties = peer.get("relationshipProperties", None) + self.log(f"Peer Relationship Properties: {peer_relationship_properties}") + if peer_properties is not None: + if peer_properties.get("class") == "PersonProperties" and peer_properties.get("typeName") == "Person": + self.peer_qname = peer_properties.get("qualifiedName") + self.peer_dname = peer_properties.get("displayName") + self.peer_fname = peer_properties.get("fullName") + self.peer_pronouns = peer_properties.get("pronouns") + # the peer's own GUID is in its elementHeader - relatedBy describes the Peer relationship + self.peer_guid = (peer.get("elementHeader") or {}).get("guid") + if peer_relationship_properties is not None: + self.peer_description = peer_relationship_properties.get("description") + self.my_network.add_row(self.peer_qname, + self.peer_dname, + self.peer_fname, + self.peer_pronouns, + self.peer_guid, + self.peer_description) + self.my_network.cursor_type = "row" + # reset variables + self.peer_qname = None + self.peer_dname = None + self.peer_fname = None + self.peer_pronouns = None + self.peer_guid = None + self.peer_description = None + continue + continue + else: + self.notify(f"No peers found for user: {self.user_name}", timeout=15, severity="error") + self.my_network.refresh() + container=self.query_one("#users_network_container", ScrollableContainer) + container.remove_children("#users_network_placeholder") + container.mount(self.my_network) + try: + self.query_one("#user_details1", Button) + except NoMatches: + self.log("Button 1 not found") + try: self.query_one("#user_details2", Button) + except NoMatches: + self.log("Button 2 not found") + container.mount(Button(label="User Details", id="user_details1", variant="primary")) + try: + self.query_one("#remove_peer_button", Button) + except NoMatches: + container.mount(Button(label="Remove from Network", id="remove_peer_button", variant="error")) + # container.refresh() + # self.my_network.refresh() + + def compose(self) -> ComposeResult: + yield Header(show_clock=True) + yield Static(self.description) + yield ScrollableContainer( + Placeholder(label="Placeholder for contacts list", id="users_network_placeholder"), + id="users_network_container") + yield ScrollableContainer( + Static("Search for user, enter the (partial) name to search for and press return:"), + Input(placeholder="Name", id="user_search_input"), + id="users_input_container") + yield ScrollableContainer(id="user_details_container") + yield Footer() + + @on(Input.Submitted, "#user_search_input") + def handle_input_submitted(self, event: Input.Submitted) -> None: + """Handle the submission of the user search input.""" + self.log(f"User search input submitted: {event.value}") + user_found = self.search_users(str(event.value)) + self.users_found.clear(columns=True) + self.users_found.add_columns("Id", "GUID", "Qualified Name") + self.users_found.cursor_type = "row" + self.users_found.zebra_stripes=True + self.notify(f"Found {len(user_found)} users with name: {event.value}", severity="information", timeout=20) + if len(user_found) == 0: + self.notify(f"No users found with name: {event.value}", severity="warning", timeout=20) + return + if isinstance(user_found, str): + self.notify(f"Error searching for users: {user_found}", severity="error", timeout=20) + return + if isinstance(user_found, list): + for user in user_found: + self.users_found.add_row(user["Display Name"], user["GUID"], user["Qualified Name"]) + self.query_one("#users_network_container", ScrollableContainer).remove_children() + self.query_one("#users_network_container", ScrollableContainer).mount(self.users_found, + Static("Please select a row to add to your network")) + elif isinstance(user_found, dict): + self.users_found.add_row(user_found["User ID"], user_found["GUID"], user_found["Qualified Name"]) + self.query_one("#users_network_container", ScrollableContainer).mount(self.users_found, + Static( + "Please select a row to add to your network")) + else: + self.log(f"Unexpected response type: {type(user_found)}") + self.notify(f"Unexpected response type: {type(user_found)}") + self.users_found.refresh() + self.query_one("#users_input_container", ScrollableContainer).remove_children() + + def search_users(self, user_name: str) -> Any: + self.log(f"Searching for users with name: {user_name}") + try: + eclient=Egeria(self.view_server, self.platform_url, self.user_name, self.user_password) + token=eclient.create_egeria_bearer_token(self.user_name, self.user_password) + user_found=eclient.find_actor_profiles( + search_string=user_name, + ignore_case=True, + output_format="DICT", + report_spec="User-Identities", + query_depth=0, + page_size=10, + ) + self.log(f"Found {user_found} users with name: {user_name}") + except PyegeriaException as e: + self.log(f"Error searching for users: {e}") + user_found="Error" + self.notify(f"Error searching for users: {e}", severity="error", timeout=20) + finally: + if eclient: + eclient.close_session() + return user_found + + @on(DataTable.RowSelected, "#users_found") + def handle_users_found_row_selected(self, event: DataTable.RowSelected) -> None: + self.log(f"User selected: {event.row_key}") + self.row_selected = event.row_key + row_data = self.users_found.get_row(self.row_selected) + self.log(f"Selected row data: {row_data}") + self.query_one("#users_input_container", ScrollableContainer).remove_children() + self.selected_user_id = row_data[0] + self.selected_GUID = row_data[1] + self.selected_qualified_name = row_data[2] + self.log(f"Selected user ID: {self.selected_user_id}") + self.log(f"Selected GUID: {self.selected_GUID}") + self.log(f"Selected qualified name: {self.selected_qualified_name}") + self.query_one("#users_input_container", ScrollableContainer).mount( + Static("Selected ID for new network contact is:"), + Static(f"{self.selected_user_id, " ", self.selected_qualified_name}"), + Static("Please provide a NickName to Store this contact under:"), + Input(placeholder="Nickname", id="nickname_input"), + Static("Optionally provide a description/comment about this relationship"), + Input(placeholder="Description", id="description_input"), + Button("Link Peer", id="link_peer_button", variant="primary") + ) + + @on(Input.Changed, "#nickname_input") + def handle_nickname_input(self, event: Input.Changed): + """Handle the submission of the user nickname input.""" + self.log(f"User nickname input submitted: {event.value}") + self.nickname = event.value + + @on(Input.Changed, "#description_input") + def handle_description_input(self, event: Input.Changed): + """Handle the submission of the user description input.""" + self.log(f"User description input submitted: {event.value}") + self.description = event.value + + @on(Button.Pressed, "#link_peer_button") + def handle_link_peer_button(self, event: Button.Pressed): + """Handle the submission of the link peer button. + Since there is no validity check and both NickName and Description are optional + we link the peer person without further upfront validation """ + #add link to peer + self.log(f"Owner GUID: {self.user_GUID} Peer selected GUID: {self.selected_GUID}") + now = datetime.now(timezone.utc).isoformat(timespec="milliseconds") + body = { + "class": "NewRelationshipRequestBody", + "properties": { + "class": "PeerProperties", + "peer_relationship_name": self.nickname, + "description": self.description, + "effective_from" : None, + "effective_to": None + } + } + self.log(f"Linking peer person: body {body} Owner GUID: {self.user_GUID} and Peer selected GUID: {self.selected_GUID}") + try: + eclient = Egeria(self.view_server, + self.platform_url, + self.user_name, + self.user_password) + token = eclient.create_egeria_bearer_token(self.user_name, self.user_password) + eclient.link_peer_person(self.user_GUID, + self.selected_GUID, + body + ) + self.notify(f"Peer person linked successfully") + except PyegeriaException as e: + self.log(f"Error linking peer person: {e}") + self.notify(f"Error linking peer person: {e}") + finally: + # clear input fields + self.query_one("#nickname_input", Input).clear() + self.query_one("#description_input", Input).clear() + # reset display to allow another search + self.query_one("#users_input_container", ScrollableContainer).remove_children() + self.query_one("#users_network_container", ScrollableContainer).remove_children() + self.query_one("#users_input_container").mount( + Static("Search for user, enter the (partial) name to search for and press return:"), + Input(placeholder="Name", id="user_search_input") + ) + self.query_one("#users_network_container").mount( + Static("Peer person network"), + self.my_network + ) + # update the data table to display the updated list of peers + # (get_peer_data also mounts the User Details button if it isn't already present) + self.get_peer_data() + return + + @on(DataTable.RowHighlighted, "#my_network") + def handle_network_member_highlighted(self,event: DataTable.RowHighlighted): + self.log(f"Row highlighted: {event.row_key}") + self.selected_network_member = event.row_key + return + + @on(DataTable.RowSelected, "#my_network") + def handle_network_member_selected(self, event: DataTable.RowSelected): + self.log(f"Row selected: {event.row_key}") + self.selected_network_member = event.row_key + return + + @on(Button.Pressed, "#user_details1") + def handle_user_details_1_button(self, event: Button.Pressed): + if self.selected_network_member: + selected_row_data = self.my_network.get_row(self.selected_network_member) + self.log(f"Selected row data: {selected_row_data}") + self.show_user_details(selected_row_data) + return + else: + self.notify(f"Please select a network member to view details before using the Button.", severity="warning", timeout=10) + return + + @on(Button.Pressed, "#remove_peer_button") + def handle_remove_peer_button(self, event: Button.Pressed): + """Remove the Peer relationship between the current user and the highlighted network member.""" + if not self.selected_network_member: + self.notify(f"Please select a network member to remove before using the Button.", severity="warning", timeout=10) + return + selected_row_data = self.my_network.get_row(self.selected_network_member) + self.log(f"Selected row data for removal: {selected_row_data}") + peer_dname = selected_row_data[1] + peer_guid = selected_row_data[4] + if not peer_guid: + self.notify(f"No GUID found for network member {peer_dname}", severity="error", timeout=10) + return + + def confirm_removal(confirmed: bool | None) -> None: + if confirmed: + self.remove_peer(peer_guid, peer_dname) + else: + self.notify(f"{peer_dname} was not removed from your network", severity="information", timeout=5) + + self.push_screen(ConfirmRemoveScreen(f"Remove {peer_dname} from your network?"), confirm_removal) + return + + def remove_peer(self, peer_guid: str, peer_dname: str) -> None: + """Unlink the peer from the current user's profile and refresh the network display.""" + body = { + "class": "DeleteRelationshipRequestBody" + } + self.log(f"Unlinking peer person: body {body} Owner GUID: {self.user_GUID} and Peer GUID: {peer_guid}") + eclient = None + try: + eclient = Egeria(self.view_server, + self.platform_url, + self.user_name, + self.user_password) + token = eclient.create_egeria_bearer_token(self.user_name, self.user_password) + # same GUID order as link_peer_person: owner first, peer second + eclient.unlink_peer_person(self.user_GUID, + peer_guid, + body + ) + self.notify(f"{peer_dname} removed from your network") + except PyegeriaException as e: + self.log(f"Error unlinking peer person: {e}") + self.notify(f"Error removing {peer_dname} from your network: {e}", severity="error", timeout=15) + return + finally: + if eclient: + eclient.close_session() + # the removed row's key is no longer valid, so clear the selection before reloading the peers + self.selected_network_member = None + # hide the details panel if it is showing the peer that was just removed + if self.details_peer_guid == peer_guid: + self.query_one("#user_details_container", ScrollableContainer).display = False + self.details_peer_guid = None + self.get_peer_data() + return + + def show_user_details(self, selected_row_data): + """ Show both contact details and community membership details for the selected network member.""" + # keep the peer's details separate from self.user_GUID, which is the logged-in user's own profile GUID + self.log(f"Selected row data: {selected_row_data}") + self.peer_qname = selected_row_data[0] + self.peer_dname = selected_row_data[1] + self.peer_fname = selected_row_data[2] + self.peer_pronouns = selected_row_data[3] + self.peer_guid = selected_row_data[4] + self.peer_description = selected_row_data[5] + self.log(f"Peer details: GUID={self.peer_guid}, qname={self.peer_qname}, full_name={self.peer_fname}, display_name={self.peer_dname}, pronouns={self.peer_pronouns}, description={self.peer_description}") + if not self.peer_guid: + self.notify(f"No GUID found for network member {self.peer_dname}", severity="error", timeout=10) + return + #clear and set up the datatables + self.my_user_contact_details.clear(columns=True) + self.my_user_contact_details.add_columns("Name", "Method Type", "Contact Type", "Service", "Value", "GUID") + self.my_user_contact_details.cursor_type = "row" + self.my_user_contact_details.zebra_stripes=True + self.my_user_communities.clear(columns=True) + self.my_user_communities.add_columns("Name", "Assignment Type", "Description", "GUID") + self.my_user_communities.cursor_type = "row" + self.my_user_communities.zebra_stripes=True + # retrieve the peer's profile and extract contact methods and communities using the + # My-User report spec (whose detail specs are My-User-Contact-Detail / My-User-Communities-Detail) + eclient = None + try: + eclient=Egeria( + self.view_server, + self.platform_url, + self.user_name, + self.user_password) + token=eclient.create_egeria_bearer_token(self.user_name, self.user_password) + peer_profile = eclient.get_actor_profile_by_guid(self.peer_guid, output_format="JSON") + peer_details = eclient._generate_my_profile_output(peer_profile, output_format="DICT", + report_spec="My-User") + self.log(f"Retrieved peer profile details: {peer_details}") + except Exception as e: + self.notify(f"Error retrieving details for {self.peer_dname}: {str(e)}", severity="error", timeout=10) + return + finally: + if eclient: + eclient.close_session() + if isinstance(peer_details, list) and peer_details: + peer_details = peer_details[0] + if not isinstance(peer_details, dict): + self.notify(f"No profile details found for {self.peer_dname}", severity="warning", timeout=10) + return + self.user_contact_details = peer_details.get("Contact Methods") or [] + self.user_communities = peer_details.get("Communities") or [] + self.log(f"User details: {self.user_contact_details}") + self.log(f"User communities: {self.user_communities}") + for contact in self.user_contact_details: + self.my_user_contact_details.add_row(contact.get("name"), + contact.get("contactMethodType"), + contact.get("contactType"), + contact.get("contactMethodService"), + contact.get("contactMethodValue"), + contact.get("guid")) + for community in self.user_communities: + self.my_user_communities.add_row(community.get("name"), + community.get("assignmentType"), + community.get("description"), + community.get("guid")) + # display the tables, mounting them the first time through + container = self.query_one("#user_details_container", ScrollableContainer) + container.display = True + self.details_peer_guid = self.peer_guid + try: + container.query_one("#user_details_heading", Static).update(f"Details for {self.peer_dname}") + except NoMatches: + container.mount(Static(f"Details for {self.peer_dname}", id="user_details_heading"), + Static("Contact Methods:"), + self.my_user_contact_details, + Static("Communities:"), + self.my_user_communities) + if not self.user_contact_details: + self.notify(f"No contact methods found for {self.peer_dname}", severity="information", timeout=10) + if not self.user_communities: + self.notify(f"No community memberships found for {self.peer_dname}", severity="information", timeout=10) + return + +def main() -> None: + """Entry point for the my_profile console script.""" + MyNetworkApp().run() + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/my_egeria/my_egeria/DemoCode/Experimental Code/related_elements_return_data.json b/my_egeria/my_egeria/DemoCode/Experimental Code/related_elements_return_data.json new file mode 100644 index 00000000..5652e33a --- /dev/null +++ b/my_egeria/my_egeria/DemoCode/Experimental Code/related_elements_return_data.json @@ -0,0 +1,252 @@ +{ + "class": "OpenMetadataRootElement", + "elementHeader": { + "class": "ElementHeader", + "headerVersion": 0, + "status": "ACTIVE", + "type": { + "typeId": "ac406bf8-e53e-49f1-9088-2af28bbbd285", + "typeName": "Person", + "superTypeNames": [ + "ActorProfile", + "Actor", + "Referenceable", + "OpenMetadataRoot" + ], + "typeVersion": 2, + "typeDescription": "An individual.", + "typeCategory": "ENTITY_DEF" + }, + "origin": { + "sourceServer": "qs-metadata-store", + "originCategory": "CONTENT_PACK", + "homeMetadataCollectionId": "655b1965-4c29-4a0e-8a5d-3f55a37b3799", + "homeMetadataCollectionName": "Coco Pharmaceuticals Combination", + "license": "Apache 2.0" + }, + "versions": { + "createdBy": "Egeria Project", + "updatedBy": "Egeria Project", + "maintainedBy": [ + "Egeria Project" + ], + "createTime": "2021-12-20T07:20:40.038+00:00", + "updateTime": "2021-12-20T07:20:40.038+00:00", + "version": 1787846084192 + }, + "guid": "8ba14282-6c51-48fa-bc68-fd241b19cad1", + "anchor": { + "class": "ElementClassification", + "headerVersion": 0, + "status": "ACTIVE", + "type": { + "typeId": "aa44f302-2e43-4669-a1e7-edaae414fc6e", + "typeName": "Anchors", + "typeVersion": 1, + "typeDescription": "Identifies the anchor entity for an element that is part of a large composite object such as an asset.", + "typeCategory": "CLASSIFICATION_DEF" + }, + "origin": { + "sourceServer": "qs-metadata-store", + "originCategory": "CONTENT_PACK", + "homeMetadataCollectionId": "655b1965-4c29-4a0e-8a5d-3f55a37b3799", + "homeMetadataCollectionName": "Coco Pharmaceuticals Combination" + }, + "versions": { + "createdBy": "Egeria Project", + "updatedBy": "Egeria Project", + "maintainedBy": [ + "Egeria Project" + ], + "createTime": "2021-12-20T07:20:40.038+00:00", + "updateTime": "2021-12-20T07:20:40.038+00:00", + "version": 1787846084192 + }, + "classificationOrigin": "ASSIGNED", + "classificationName": "Anchors", + "classificationProperties": { + "class": "AnchorsProperties", + "typeName": "Anchors", + "anchorTypeName": "Person", + "anchorDomainName": "Actor" + } + } + }, + "contactDetails":[ + { + "class": "RelatedMetadataElementSummary", + "relationshipHeader": { + "class": "ElementHeader", + "headerVersion": 0, + "status": "ACTIVE", + "type": { + "typeId": "6cb9af43-184e-4dfa-854a-1572bcf0fe75", + "typeName": "ContactThrough", + "typeVersion": 2, + "typeDescription": "he contact details associated with an actor profile.", + "typeCategory": "RELATIONSHIP_DEF" + }, + "origin": { + "sourceServer": "qs-metadata-store", + "originCategory": "CONTENT_PACK", + "homeMetadataCollectionId": "655b1965-4c29-4a0e-8a5d-3f55a37b3799", + "homeMetadataCollectionName": "Coco Pharmaceuticals Combination", + "license": "Apache 2.0" + }, + "versions": { + "createdBy": "Egeria Project", + "updatedBy": "Egeria Project", + "maintainedBy": [ + "Egeria Project" + ], + "createTime": "2021-12-20T07:20:40.038+00:00", + "updateTime": "2021-12-20T07:20:40.038+00:00", + "version": 1787846084192 + }, + "guid": "9758d8cb-e9a0-481e-aa5c-1aa21459d6dc" + }, + "relatedElement": { + "elementHeader": { + "class": "ElementHeader", + "headerVersion": 0, + "status": "ACTIVE", + "type": { + "typeId": "79296df8-645a-4ef7-a011-912d1cdcf75a", + "typeName": "ContactDetails", + "superTypeNames": [ + "Referenceable", + "OpenMetadataRoot" + ], + "typeVersion": 1, + "typeDescription": "Information on how to send a message to an individual or automated process.", + "typeCategory": "ENTITY_DEF" + }, + "origin": { + "sourceServer": "qs-metadata-store", + "originCategory": "CONTENT_PACK", + "homeMetadataCollectionId": "655b1965-4c29-4a0e-8a5d-3f55a37b3799", + "homeMetadataCollectionName": "Coco Pharmaceuticals Combination", + "license": "Apache 2.0" + }, + "versions": { + "createdBy": "Egeria Project", + "updatedBy": "Egeria Project", + "maintainedBy": [ + "Egeria Project" + ], + "createTime": "2021-12-20T07:20:40.038+00:00", + "updateTime": "2021-12-20T07:20:40.038+00:00", + "version": 1787846084192 + }, + "guid": "1e9e83c8-7d70-42f6-a49c-0b093ccdb807", + "anchor": { + "class": "ElementClassification", + "headerVersion": 0, + "status": "ACTIVE", + "type": { + "typeId": "aa44f302-2e43-4669-a1e7-edaae414fc6e", + "typeName": "Anchors", + "typeVersion": 1, + "typeDescription": "Identifies the anchor entity for an element that is part of a large composite object such as an asset.", + "typeCategory": "CLASSIFICATION_DEF" + }, + "origin": { + "sourceServer": "qs-metadata-store", + "originCategory": "CONTENT_PACK", + "homeMetadataCollectionId": "655b1965-4c29-4a0e-8a5d-3f55a37b3799", + "homeMetadataCollectionName": "Coco Pharmaceuticals Combination" + }, + "versions": { + "createdBy": "Egeria Project", + "updatedBy": "Egeria Project", + "maintainedBy": [ + "Egeria Project" + ], + "createTime": "2021-12-20T07:20:40.038+00:00", + "updateTime": "2021-12-20T07:20:40.038+00:00", + "version": 1787846084192 + }, + "classificationOrigin": "ASSIGNED", + "classificationName": "Anchors", + "classificationProperties": { + "class": "AnchorsProperties", + "typeName": "Anchors", + "anchorGUID": "8ba14282-6c51-48fa-bc68-fd241b19cad1", + "anchorTypeName": "Person", + "anchorDomainName": "Actor" + } + } + }, + "properties": { + "class": "ContactDetailsProperties", + "typeName": "ContactDetails", + "displayName": "Work Email Address", + "contactType": "Work Email", + "contactMethodType": "EMAIL", + "contactMethodService": "Email address assigned by Coco Pharmaceuticals.", + "contactMethodValue": "TessaTube@Coco-Pharmaceuticals.org" + } + }, + "relatedElementAtEnd1": false + } + ], + "relatedBy": { + "relationshipHeader": { + "class": "ElementHeader", + "headerVersion": 0, + "status": "ACTIVE", + "type": { + "typeId": "4a316abe-bccd-4d11-ad5a-4bfb4079b80b", + "typeName": "Peer", + "typeVersion": 2, + "typeDescription": "Relationship identifying a person's peer network.", + "typeCategory": "RELATIONSHIP_DEF" + }, + "origin": { + "sourceServer": "qs-metadata-store", + "originCategory": "LOCAL_COHORT", + "homeMetadataCollectionId": "9905c3cb-94c5-4494-9229-0d6f69c0b842", + "homeMetadataCollectionName": "qs-metadata-store" + }, + "versions": { + "createdBy": "tessatube", + "updatedBy": "tessatube", + "maintainedBy": [ + "tessatube" + ], + "createTime": "2026-10-06T17:42:03.814+00:00", + "updateTime": "2026-10-06T17:42:03.814+00:00", + "version": 1, + "lastRequestId": "4a59ea13-a10f-419d-a0d2-8ad50cdc94aa" + }, + "guid": "0e9c5f02-8797-4716-ba2d-945256092924" + }, + "relationshipProperties": { + "class": "PeerProperties", + "typeName": "Peer", + "description": "Testing Peers" + } + }, + "properties": { + "class": "PersonProperties", + "typeName": "Person", + "qualifiedName": "Person::USA::302145", + "displayName": "Tessa Tube", + "additionalProperties": { + "workLocation": "3", + "countryCode": "USA", + "organization": "Coco Pharmaceuticals Ltd", + "jobLevel": "7" + }, + "courtesyTitle": "Dr", + "initials": "T", + "givenNames": "Tessa", + "surname": "Tube", + "fullName": "Dr Tessa Tube", + "pronouns": "She/her/hers", + "jobTitle": "Lead Researcher and head of Salk Research Lab", + "employeeNumber": "302145", + "employeeType": "FTP" + }, + "mermaidGraph": "---\ntitle: Person - Tessa Tube [8ba14282-6c51-48fa-bc68-fd241b19cad1]\n---\nflowchart TD\n%%{init: {\"flowchart\": {\"htmlLabels\": false}} }%%\n\n1@{ shape: rounded, label: \"*Person*\n**Tessa Tube**\"}\n2@{ shape: rounded, label: \"*Contact Details*\n**Work Email Address**\"}\n1==>|\"Contact Through\"|2\nstyle 1 color:#000000, fill:#e0ab18, stroke:#004563\nstyle 2 color:#000000, fill:#39add1, stroke:#004563\n" + } \ No newline at end of file diff --git a/my_egeria/my_egeria/DemoCode/Journals_Demo/__init__.py b/my_egeria/my_egeria/DemoCode/Journals_Demo/__init__.py deleted file mode 100644 index d7c7a021..00000000 --- a/my_egeria/my_egeria/DemoCode/Journals_Demo/__init__.py +++ /dev/null @@ -1,7 +0,0 @@ -""" - PDX-License-Identifier: Apache-2.0 - Copyright Contributors to the ODPi Egeria project. - - This file provides a set of report specification related functions for my_egeria. - -""" \ No newline at end of file diff --git a/my_egeria/my_egeria/DemoCode/My_Profile/MainScreen.py b/my_egeria/my_egeria/DemoCode/My_Profile/MainScreen.py index 4184fe39..9b183932 100644 --- a/my_egeria/my_egeria/DemoCode/My_Profile/MainScreen.py +++ b/my_egeria/my_egeria/DemoCode/My_Profile/MainScreen.py @@ -60,6 +60,7 @@ def compose(self) -> ComposeResult: Option("[] Subscriptions"), Option("[] Technology Types"), Option("[] User Bookmarks"), + Option("[] My Network"), Option("[] Leave Feedback"), Option("[] Feedback Log"), id="other_function_list" diff --git a/my_egeria/my_egeria/DemoCode/My_Profile/MyNetworkScreen.py b/my_egeria/my_egeria/DemoCode/My_Profile/MyNetworkScreen.py new file mode 100644 index 00000000..b2f42a50 --- /dev/null +++ b/my_egeria/my_egeria/DemoCode/My_Profile/MyNetworkScreen.py @@ -0,0 +1,272 @@ +""" + PDX-License-Identifier: Apache-2.0 + Copyright Contributors to the ODPi Egeria project. + + This file provides the My Network screen for My Profile. + +""" +from textual import on +from textual.app import ComposeResult +from textual.containers import ScrollableContainer, Container, Horizontal +from textual.screen import ModalScreen +from textual.widgets import DataTable, Header, Static, Footer, Input, Button + + +class ConfirmRemovePeerScreen(ModalScreen[bool]): + """Confirmation prompt shown before a person is removed from the user's network. + + Dismisses with True to proceed with the removal, False to cancel. + """ + + BINDINGS = [("escape", "cancel", "Cancel")] + + CSS_PATH = "my_profile.tcss" + + def __init__(self, peer_name: str, *args, **kwargs): + super().__init__(*args, **kwargs, id="confirm_remove_peer_screen") + self.peer_name = peer_name + + def compose(self) -> ComposeResult: + yield Header(show_clock=True) + yield Static("Confirm Removal", classes="span-3", id="confirm_remove_peer_title") + yield ScrollableContainer( + Static(f"Remove {self.peer_name} from your network?"), + Static("Their profile is not changed, only the link between you."), + Horizontal( + Button("Remove", id="confirm_remove_peer_btn", variant="error"), + Button("Cancel", id="cancel_remove_peer_btn", variant="primary"), + ), + id="confirm_remove_peer_container", + ) + yield Footer() + + def on_mount(self) -> None: + self.title = "Egeria - my_profile" + self.sub_title = "Confirm Removal - My Network" + + @on(Button.Pressed, "#confirm_remove_peer_btn") + def handle_confirm_button(self, event: Button.Pressed) -> None: + self.dismiss(True) + + @on(Button.Pressed, "#cancel_remove_peer_btn") + def handle_cancel_button(self, event: Button.Pressed) -> None: + self.dismiss(False) + + def action_cancel(self) -> None: + self.dismiss(False) + + +class MyNetworkScreen(ModalScreen): + """List the current user's network of peers, add people to it and remove them. + + The Egeria work is done by the app's NetworkMixin (list_my_peers, search_people, + link_peer, unlink_peer, get_peer_details); this screen only drives it. + """ + + BINDINGS = [ + ("q", "quit", "Quit"), + ("v", "view_details", "View Highlighted Person"), + ("d", "remove_peer", "Remove Highlighted Person"), + ] + + CSS_PATH = "my_profile.tcss" + + def __init__(self, my_peers, *args, **kwargs): + super().__init__(id="my_network_screen", *args, **kwargs) + self.title = "Egeria" + self.sub_title = "My Network" + # Rows of (qualified name, display name, full name, pronouns, GUID, description) + self.my_peers_data = list(my_peers or []) + # (display name, GUID) of the person picked from the search results to add + self.selected_person = None + # GUID of the peer whose details are currently displayed + self.details_peer_guid = None + + def compose(self) -> ComposeResult: + yield Header(show_clock=True) + yield Static("My Network", classes="span-3", id="my_network_title") + yield ScrollableContainer( + Static("The people in your network. 'v' shows the highlighted person's details; 'd' removes them."), + DataTable(id="my_network_table", zebra_stripes=True, cursor_type="row"), + Horizontal( + Button("Person Details", id="peer_details_btn", variant="primary"), + Button("Remove from Network", id="remove_peer_btn", variant="error"), + id="my_network_buttons", + ), + id="my_network_container", + ) + yield ScrollableContainer( + Static("Add to your network: enter a (partial) name and press return"), + Input(placeholder="Name", id="people_search_input"), + DataTable(id="people_found_table", zebra_stripes=True, cursor_type="row"), + Container(id="add_peer_container"), + id="people_search_container", + ) + yield ScrollableContainer( + Static("", id="peer_details_heading"), + Static("Contact Methods:"), + DataTable(id="peer_contacts_table", zebra_stripes=True, cursor_type="row"), + Static("Communities:"), + DataTable(id="peer_communities_table", zebra_stripes=True, cursor_type="row"), + id="peer_details_container", + ) + yield Footer() + + def on_mount(self) -> None: + self.query_one("#my_network_table", DataTable).add_columns( + "Qualified Name", "Display Name", "Full Name", "Pronouns", "GUID", "Description") + self.query_one("#people_found_table", DataTable).add_columns("Name", "GUID", "Qualified Name") + self.query_one("#peer_contacts_table", DataTable).add_columns( + "Name", "Method Type", "Contact Type", "Service", "Value", "GUID") + self.query_one("#peer_communities_table", DataTable).add_columns( + "Name", "Assignment Type", "Description", "GUID") + self.query_one("#peer_details_container").display = False + self._fill_table() + + def _fill_table(self) -> None: + table = self.query_one("#my_network_table", DataTable) + table.clear() + for row in self.my_peers_data: + table.add_row(*row, key=row[4]) + if not self.my_peers_data: + self.notify("There is no-one in your network yet - search for someone to add them", timeout=5) + + def _reload(self) -> None: + rows = self.app.list_my_peers() + if rows is not None: + self.my_peers_data = rows + self._fill_table() + + def _highlighted_peer(self) -> tuple[str, str] | None: + """(display name, GUID) of the highlighted network row, or None if the network is empty.""" + table = self.query_one("#my_network_table", DataTable) + if table.row_count == 0: + self.notify("There is no-one in your network to select", timeout=5, severity="warning") + return None + row = table.get_row_at(table.cursor_row) + return row[1] or row[0], row[4] + + def action_quit(self) -> None: + """ The user elects to quit the network function """ + self.dismiss(200) + + @on(Button.Pressed, "#peer_details_btn") + def handle_peer_details_button(self, event: Button.Pressed) -> None: + self.action_view_details() + + def action_view_details(self) -> None: + """ Show the contact methods and communities of the highlighted peer """ + peer = self._highlighted_peer() + if not peer: + return + peer_name, peer_guid = peer + details = self.app.get_peer_details(peer_guid) + if details is None: + return + contacts, communities = details + contacts_table = self.query_one("#peer_contacts_table", DataTable) + contacts_table.clear() + for contact in contacts: + contacts_table.add_row(contact.get("name"), + contact.get("contactMethodType"), + contact.get("contactType"), + contact.get("contactMethodService"), + contact.get("contactMethodValue"), + contact.get("guid")) + communities_table = self.query_one("#peer_communities_table", DataTable) + communities_table.clear() + for community in communities: + communities_table.add_row(community.get("name"), + community.get("assignmentType"), + community.get("description"), + community.get("guid")) + self.query_one("#peer_details_heading", Static).update(f"[b]Details for {peer_name}[/b]") + self.query_one("#peer_details_container").display = True + self.details_peer_guid = peer_guid + if not contacts: + self.notify(f"No contact methods found for {peer_name}", timeout=10) + if not communities: + self.notify(f"No community memberships found for {peer_name}", timeout=10) + + @on(Button.Pressed, "#remove_peer_btn") + def handle_remove_peer_button(self, event: Button.Pressed) -> None: + self.action_remove_peer() + + def action_remove_peer(self) -> None: + """ The user wants to remove the highlighted peer, after confirmation """ + peer = self._highlighted_peer() + if not peer: + return + peer_name, peer_guid = peer + + def on_confirm(confirmed: bool | None) -> None: + if not confirmed: + self.notify(f"{peer_name} was not removed from your network", timeout=5) + return + if self.app.unlink_peer(peer_guid): + # hide the details panel if it is showing the person just removed + if self.details_peer_guid == peer_guid: + self.query_one("#peer_details_container").display = False + self.details_peer_guid = None + self._reload() + + self.app.push_screen(ConfirmRemovePeerScreen(peer_name), callback=on_confirm) + + @on(Input.Submitted, "#people_search_input") + def handle_people_search(self, event: Input.Submitted) -> None: + search_string = event.value.strip() + if not search_string: + self.notify("Enter a (partial) name to search for", timeout=5, severity="warning") + return + found = self.app.search_people(search_string) + if found is None: + return + table = self.query_one("#people_found_table", DataTable) + table.clear() + in_network = {row[4] for row in self.my_peers_data} + for name, guid, qualified_name in found: + if guid not in in_network: + table.add_row(name, guid, qualified_name, key=guid) + if table.row_count == 0: + self.notify(f"No-one new found matching {search_string}", timeout=5, severity="warning") + else: + self.notify("Select a person to add them to your network", timeout=5) + table.focus() + + @on(DataTable.RowSelected, "#people_found_table") + def handle_person_selected(self, event: DataTable.RowSelected) -> None: + name, guid, qualified_name = event.data_table.get_row(event.row_key) + self.selected_person = (name, guid) + container = self.query_one("#add_peer_container", Container) + container.remove_children() + container.mount( + Static(f"Add [b]{name}[/b] ({qualified_name}) to your network"), + Input(placeholder="Nickname (optional)", id="peer_nickname_input"), + Input(placeholder="Description/comment about this relationship (optional)", id="peer_description_input"), + Horizontal( + Button("Add to Network", id="add_peer_btn", variant="primary"), + Button("Cancel", id="cancel_add_peer_btn", variant="warning"), + ), + ) + self.query_one("#peer_nickname_input", Input).focus() + + @on(Button.Pressed, "#add_peer_btn") + def handle_add_peer_button(self, event: Button.Pressed) -> None: + if not self.selected_person: + return + nickname = self.query_one("#peer_nickname_input", Input).value.strip() + description = self.query_one("#peer_description_input", Input).value.strip() + if self.app.link_peer(self.selected_person[1], nickname, description): + self._clear_search() + self._reload() + + @on(Button.Pressed, "#cancel_add_peer_btn") + def handle_cancel_add_peer_button(self, event: Button.Pressed) -> None: + self.query_one("#add_peer_container", Container).remove_children() + self.selected_person = None + + def _clear_search(self) -> None: + self.query_one("#add_peer_container", Container).remove_children() + self.query_one("#people_found_table", DataTable).clear() + self.query_one("#people_search_input", Input).clear() + self.selected_person = None diff --git a/my_egeria/my_egeria/DemoCode/My_Profile/ShopForDataScreen.py b/my_egeria/my_egeria/DemoCode/My_Profile/ShopForDataScreen.py index 529cf4ce..b99c2937 100644 --- a/my_egeria/my_egeria/DemoCode/My_Profile/ShopForDataScreen.py +++ b/my_egeria/my_egeria/DemoCode/My_Profile/ShopForDataScreen.py @@ -11,22 +11,24 @@ from pyegeria import load_app_config, settings from textual import on from textual.app import ComposeResult +from textual.binding import Binding from textual.containers import ScrollableContainer from textual.events import Action from textual.screen import Screen -from textual.widgets import DataTable, Header, TextArea, Static, Footer +from textual.widgets import DataTable, Header, Static, Footer class ShopForDataScreen(Screen): """ Screen to Present a choice of different data sources to the user.""" BINDINGS = [("q", "dismiss(200)", "Quit"), + ("b", "back", "Go back"), + Binding("escape", "back", "Go back", show=False), ("s", "sample_data_source", "Sample data source"), ("u", "subscribe_to_data_source", "Subscribe"), ("ctrl+s", "subscribe_to_data_source", "Subscribe"), ("t", "search_for_term", "Search Glossary Terms"), ("k", "bookmark_row", "Bookmark"), - ("ctrl+b", "bookmarks", "Bookmarks"), - ("b", "back", "Go back")] + ("ctrl+b", "bookmarks", "Bookmarks")] CSS_PATH = "my_profile.tcss" @@ -86,7 +88,8 @@ def __init__( def compose(self) -> ComposeResult: """ Compose the UI components for the ShopForDataScreen screen.""" yield Header(show_clock=True) - yield TextArea(f"\n\n\nPlease click on a line to select a data source:") + # A Static, not a TextArea: a focused TextArea swallows the screen's letter key bindings (b, q, s, ...) + yield Static("\nPlease click on a line to select a data source, or press b or Escape to go back:") yield ScrollableContainer( Static("Glossary"), self.glossary_table) 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 5793f19d..705c8252 100644 --- a/my_egeria/my_egeria/DemoCode/My_Profile/my_profile.tcss +++ b/my_egeria/my_egeria/DemoCode/My_Profile/my_profile.tcss @@ -62,11 +62,52 @@ FeedbackScreen, FeedbackLogScreen { background: $background; } +/* Title row, then network + search side by side, then the selected peer's details */ +MyNetworkScreen { + grid-rows: auto 1fr 1fr; + background: $background; +} + +#my_network_container { + column-span: 2; + width: 100%; + height: 100%; +} + +#people_search_container { + width: 100%; + height: 100%; +} + +#peer_details_container { + column-span: 3; + width: 100%; + height: 100%; +} + +#my_network_table { + width: 100%; + height: 1fr; +} + +#my_network_buttons, +#add_peer_container, +#add_peer_container Horizontal { + height: auto; +} + +#people_found_table, +#peer_contacts_table, +#peer_communities_table { + height: auto; + max-height: 12; +} + ModalScreen > Header, ModalScreen > Footer { column-span: 3; } -#edit_profile_title, #create_profile_title, #edit_teams_title, #edit_roles_title, #edit_communities_title, #edit_identities_title, #status_title, #generic_data_view_title, #feedback_title, #feedback_log_title, .span-3 { +#my_network_title, #confirm_remove_peer_title, #edit_profile_title, #create_profile_title, #edit_teams_title, #edit_roles_title, #edit_communities_title, #edit_identities_title, #status_title, #generic_data_view_title, #feedback_title, #feedback_log_title, .span-3 { column-span: 3; text-align: center; background: darkblue; @@ -96,6 +137,7 @@ ModalScreen > Header, ModalScreen > Footer { #edit_projects_container, #edit_todos_container, #confirm_delete_container, +#confirm_remove_peer_container, #status_container, #user_identities_container, #generic_data_view_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 17fea454..2b3674bf 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 @@ -88,10 +88,11 @@ from feedback_handler import FeedbackMixin, FEEDBACK_LOG_OWNER from bookmarks_handler import BookmarksMixin, bookmarks_qualified_name from MyBookMarksScreen import MyBookMarksScreen +from network_handler import NetworkMixin class MyProfileApp(App, TechTypesMixin, ShopForDataMixin, TeamRolesMixin, ElementsCrudMixin, FeedbackMixin, - BookmarksMixin): + BookmarksMixin, NetworkMixin): """My Profile App. Retrieves a user's profile from Egeria and displays current work items. @@ -678,6 +679,8 @@ async def handle_option_selected(self, event: OptionList.OptionSelected) -> None ) elif selected_option == "User Bookmarks": self.show_my_bookmarks() + elif selected_option == "My Network": + self.show_my_network() 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/network_handler.py b/my_egeria/my_egeria/DemoCode/My_Profile/network_handler.py new file mode 100644 index 00000000..aa1a27d8 --- /dev/null +++ b/my_egeria/my_egeria/DemoCode/My_Profile/network_handler.py @@ -0,0 +1,186 @@ +""" + PDX-License-Identifier: Apache-2.0 + Copyright Contributors to the ODPi Egeria project. + + My Network handler mixin for My Profile Textual App. + + A user's network is the set of people linked to their profile by a Peer + relationship. Peers are found with find_actor_profiles, linked with + link_peer_person and removed with unlink_peer_person. +""" + +import sys +from pathlib import Path +from typing import Any + +# Ensure project root is on sys.path +root_path = Path(__file__).resolve().parents[4] +if str(root_path) not in sys.path: + sys.path.append(str(root_path)) + +current_dir = Path(__file__).resolve().parent +if str(current_dir) not in sys.path: + sys.path.append(str(current_dir)) + +from pyegeria import Egeria, PyegeriaException +from MyNetworkScreen import MyNetworkScreen + +PEER_RELATIONSHIP = "Peer" + + +class NetworkMixin: + """Mixin class providing the My Network feature for MyProfileApp.""" + + def _network_client(self) -> Egeria: + client = Egeria(view_server=self.view_server, platform_url=self.platform_url, + user_id=self.user_name, user_pwd=self.user_password) + client.create_egeria_bearer_token(self.user_name, self.user_password) + return client + + def list_my_peers(self) -> list[tuple[str, str, str, str, str, str]] | None: + """The user's peers as (qualified name, display name, full name, pronouns, GUID, description) rows, + or None if Egeria could not be reached.""" + client = None + try: + client = self._network_client() + body = { + "class": "ResultsRequestBody", + "metadataElementTypeName": "Referenceable", + "includeOnlyRelationships": ["ContactThrough"], + } + peers = client.get_related_elements(element_guid=self.user_GUID, + relationship_type=PEER_RELATIONSHIP, + body=body) + except PyegeriaException as e: + self.log(f"Retrieving peers failed: {e}") + self.notify(f"Could not retrieve your network: {e}", timeout=10, severity="error") + return None + finally: + if client: + client.close_session() + self.log(f"Peers: {peers}") + rows = [] + # "No elements found" is returned, rather than an exception, when the user has no peers + for peer in peers if isinstance(peers, list) else []: + properties = peer.get("properties") or {} + if properties.get("class") != "PersonProperties": + continue + # the peer's own GUID is in its elementHeader - relatedBy describes the Peer relationship + guid = (peer.get("elementHeader") or {}).get("guid") + if not guid: + continue + relationship_properties = peer.get("relationshipProperties") or {} + rows.append((properties.get("qualifiedName") or "", + properties.get("displayName") or "", + properties.get("fullName") or "", + properties.get("pronouns") or "", + guid, + relationship_properties.get("description") or "")) + return rows + + def show_my_network(self) -> None: + """Show the My Network screen for the current user.""" + if not getattr(self, "user_GUID", ""): + self.notify("Your profile has not loaded yet, so your network can't be shown", timeout=10, + severity="warning") + return + peers = self.list_my_peers() + if peers is None: + return + self.push_screen(MyNetworkScreen(peers), callback=self.my_network_callback) + + def my_network_callback(self, result: Any) -> None: + self.log(f"My Network screen returned: {result}") + + def search_people(self, search_string: str) -> list[tuple[str, str, str]] | None: + """People whose profile matches search_string, as (display name, GUID, qualified name) rows, + or None if Egeria could not be reached.""" + client = None + try: + client = self._network_client() + found = client.find_actor_profiles(search_string=search_string, ignore_case=True, + output_format="DICT", report_spec="User-Identities", + query_depth=0, page_size=10) + except PyegeriaException as e: + self.log(f"Searching for {search_string} failed: {e}") + self.notify(f"Searching for people failed: {e}", timeout=10, severity="error") + return None + finally: + if client: + client.close_session() + self.log(f"People found for {search_string}: {found}") + if isinstance(found, dict): + found = [found] + rows = [] + for person in found if isinstance(found, list) else []: + guid = person.get("GUID") + # the current user can't be their own peer + if guid and guid != self.user_GUID: + rows.append((person.get("Display Name") or "", guid, person.get("Qualified Name") or "")) + return rows + + def link_peer(self, peer_guid: str, nickname: str = "", description: str = "") -> bool: + """Add the person with the given GUID to the user's network. Returns True on success.""" + body = { + "class": "NewRelationshipRequestBody", + "properties": { + "class": "PeerProperties", + "peer_relationship_name": nickname, + "description": description, + "effective_from": None, + "effective_to": None, + } + } + client = None + try: + client = self._network_client() + client.link_peer_person(self.user_GUID, peer_guid, body) + self.notify("Added to your network", timeout=10, severity="information") + return True + except PyegeriaException as e: + self.log(f"Linking peer {peer_guid} failed: {e}") + self.notify(f"Adding to your network failed: {e}", timeout=10, severity="error") + return False + finally: + if client: + client.close_session() + + def unlink_peer(self, peer_guid: str) -> bool: + """Remove the person with the given GUID from the user's network. Returns True on success.""" + client = None + try: + client = self._network_client() + # same GUID order as link_peer_person: owner first, peer second + client.unlink_peer_person(self.user_GUID, peer_guid, {"class": "DeleteRelationshipRequestBody"}) + self.notify("Removed from your network", timeout=10, severity="information") + return True + except PyegeriaException as e: + self.log(f"Unlinking peer {peer_guid} failed: {e}") + self.notify(f"Removing from your network failed: {e}", timeout=10, severity="error") + return False + finally: + if client: + client.close_session() + + def get_peer_details(self, peer_guid: str) -> tuple[list[dict], list[dict]] | None: + """The peer's (contact methods, communities), or None if they could not be retrieved.""" + client = None + try: + client = self._network_client() + profile = client.get_actor_profile_by_guid(peer_guid, output_format="JSON") + # My-User's detail specs are My-User-Contact-Detail / My-User-Communities-Detail + details = client._generate_my_profile_output(profile, output_format="DICT", report_spec="My-User") + except PyegeriaException as e: + self.log(f"Retrieving details for peer {peer_guid} failed: {e}") + self.notify(f"Could not retrieve the details: {e}", timeout=10, severity="error") + return None + finally: + if client: + client.close_session() + self.log(f"Peer details: {details}") + if isinstance(details, list) and details: + details = details[0] + if not isinstance(details, dict): + self.notify("No profile details found", timeout=10, severity="warning") + return None + return details.get("Contact Methods") or [], details.get("Communities") or [] 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 b4a66bb9..0b2284f4 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 @@ -20,7 +20,7 @@ if str(current_dir) not in sys.path: sys.path.append(str(current_dir)) -from pyegeria import AutomatedCuration, PyegeriaException, print_basic_exception, load_app_config, settings +from pyegeria import AutomatedCuration, PyegeriaException, print_basic_exception, load_app_config, settings, Egeria from StatusScreen import StatusScreen from TechnologyTypeScreens import ( TechnologyTypesScreen, @@ -64,6 +64,7 @@ async def tech_type_callback(self, result: Any) -> int: If the result is int (4xx) it indicates an error in the screen. If the result is str it contains the GUID of the selected technology type. """ + self.result = result # clear local data fields tech_type_description = "" self.tech_type_data = {} @@ -71,15 +72,21 @@ async def tech_type_callback(self, result: Any) -> int: self.tech_type_templates = [] self.tech_type_processes = [] # check that we got a valid result from the screen and process accordingly - if not result or isinstance(result, int): + if not self.result or isinstance(self.result, int): self.log(f"Technology Types screen cancelled or failed; return: {result}.") self._show_main_screen() - return result + return self.result + # The screen dismisses quit/go back as the strings "200"/"201", not ints + if str(self.result) in ("200", "201"): + self.log(f"Technology Types Screen returned on a quit/go back directive: {self.result}") + self._show_main_screen() + return self.result + self.selected_t_node = str(result) self.log(f"Technology Types screen returned: {self.selected_t_node}") # Request details for selected tech type try: - p_client = AutomatedCuration(self.view_server, self.platform_url, self.user_name, self.user_password) + p_client = Egeria(self.view_server, self.platform_url, self.user_name, self.user_password) token = p_client.create_egeria_bearer_token(self.user_name, self.user_password) self.tech_type_data = p_client.get_tech_type_detail( filter_string=self.selected_t_node, @@ -91,8 +98,15 @@ async def tech_type_callback(self, result: Any) -> int: error_message = f"Failed to retrieve technology type process details for {getattr(self, 'selected_q_name', self.selected_t_node)}: {str(e)}" self.log(f"Error retrieving technology type details: {error_category}, {error_message}") await self.push_screen(StatusScreen(f"{error_category}: {error_message}"), callback=self.status_callback) + return 417 self.log(f"Technology Type Data: {self.tech_type_data}, type: {type(self.tech_type_data)}") + # Egeria returns a plain string such as "No elements found" when there is no detail + if not isinstance(self.tech_type_data, dict): + self.notify(f"No details found for technology type '{self.selected_t_node}'", + timeout=10, severity="warning") + self._show_main_screen() + return 200 if "specificationMermaidGraph" in self.tech_type_data: del self.tech_type_data["specificationMermaidGraph"] diff --git a/my_report_specs/my_profile_report_specs.json b/my_report_specs/my_profile_report_specs.json new file mode 100644 index 00000000..cb88fd3d --- /dev/null +++ b/my_report_specs/my_profile_report_specs.json @@ -0,0 +1,34 @@ +{ + "My-User-Peers": { + "target_type": "Peer", + "heading": "Peers", + "description": "Detailed Peer Connections", + "family": "MyProfile", + "formats": [ + { + "types": ["LIST", "REPORT", "DICT", "TABLE"], + "attributes": [ + { + "name": "Peer", + "key": "relatedBy.relationshipHeader.type.typeName" + }, + { + "name": "guid", + "key": "elementHeader.guid" + }, + { + "name": "peerProperties", + "key": "relatedBy.relationshipProperties.class" + } + ] + } + ], + "action": { + "function": "ClassificationExplorer.get_related_elements", + "required_params": ["element_guid"], + "spec_params": { + "relationship_type": "Peer" + } + } + } +} 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 caf2f476..ce5d9b6d 100644 --- a/tests/micro-tests/my_profile/test_my_profile_app.py +++ b/tests/micro-tests/my_profile/test_my_profile_app.py @@ -256,6 +256,15 @@ async def test_handle_option_selected(self, option_name, handler_attr, screen_cl call_args = app.push_screen.call_args[0] assert isinstance(call_args[0], screen_cls) + @pytest.mark.asyncio + async def test_my_network_option_shows_network(self): + app = MyProfileApp() + app.show_my_network = MagicMock() + event = MagicMock() + event.option.prompt = "[] My Network" + await app.handle_option_selected(event) + app.show_my_network.assert_called_once_with() + def test_status_callback(self): """Closing a status screen returns to the main screen; it must not end the app.""" app = MyProfileApp() diff --git a/tests/micro-tests/my_profile/test_network_handler.py b/tests/micro-tests/my_profile/test_network_handler.py new file mode 100644 index 00000000..04b78245 --- /dev/null +++ b/tests/micro-tests/my_profile/test_network_handler.py @@ -0,0 +1,267 @@ +""" + PDX-License-Identifier: Apache-2.0 + Copyright Contributors to the ODPi Egeria project. + + Unit tests for the My Network mixin and screen (a user's network is the people + linked to their profile by a Peer relationship). +""" + +from unittest.mock import MagicMock, patch + +import pytest +from textual.app import App +from textual.widgets import DataTable, Input, Button + +from network_handler import NetworkMixin +from MyNetworkScreen import MyNetworkScreen, ConfirmRemovePeerScreen +from pyegeria import PyegeriaException + + +def _peer(guid, display_name, description="", type_class="PersonProperties"): + return {"elementHeader": {"guid": guid}, + "properties": {"class": type_class, "qualifiedName": f"Person::{guid}", "displayName": display_name, + "fullName": f"{display_name} Full", "pronouns": "they/them"}, + "relatedBy": {"relationshipHeader": {"guid": f"rel-{guid}"}}, + "relationshipProperties": {"description": description}} + + +class DummyApp(NetworkMixin): + def __init__(self): + self.view_server = "qs-view-server" + self.platform_url = "https://127.0.0.1:9443" + self.user_name = "garygeeke" + self.user_password = "secret" + self.user_GUID = "profile-guid" + self.notices = [] + self.pushed = [] + + def log(self, *args, **kwargs): + pass + + def notify(self, message, **kwargs): + self.notices.append((message, kwargs.get("severity"))) + + def push_screen(self, screen, callback=None): + self.pushed.append((screen, callback)) + + +@pytest.fixture +def client(): + with patch("network_handler.Egeria") as mock_egeria: + yield mock_egeria.return_value + + +class TestNetworkMixin: + def test_list_returns_peer_rows_with_person_guid(self, client): + client.get_related_elements.return_value = [ + _peer("erin-guid", "Erin", "Governance buddy"), + _peer("thing-guid", "Not a person", type_class="AssetProperties"), + ] + rows = DummyApp().list_my_peers() + assert client.get_related_elements.call_args.kwargs["element_guid"] == "profile-guid" + assert client.get_related_elements.call_args.kwargs["relationship_type"] == "Peer" + # The person's own GUID, not the relationship's; non-people are skipped + assert rows == [("Person::erin-guid", "Erin", "Erin Full", "they/them", "erin-guid", "Governance buddy")] + client.close_session.assert_called_once() + + def test_list_without_peers_is_empty(self, client): + client.get_related_elements.return_value = "No elements found" + assert DummyApp().list_my_peers() == [] + + def test_list_error_returns_none(self, client): + client.get_related_elements.side_effect = PyegeriaException("boom") + app = DummyApp() + assert app.list_my_peers() is None + assert app.notices[-1][1] == "error" + + def test_show_pushes_screen(self, client): + client.get_related_elements.return_value = [_peer("erin-guid", "Erin")] + app = DummyApp() + app.show_my_network() + screen, callback = app.pushed[-1] + assert isinstance(screen, MyNetworkScreen) + assert screen.my_peers_data[0][4] == "erin-guid" + + def test_show_without_profile_is_refused(self, client): + app = DummyApp() + app.user_GUID = "" + app.show_my_network() + assert app.pushed == [] + client.get_related_elements.assert_not_called() + assert app.notices[-1][1] == "warning" + + def test_search_excludes_current_user(self, client): + client.find_actor_profiles.return_value = [ + {"Display Name": "Tanya", "GUID": "tanya-guid", "Qualified Name": "Person::tanya"}, + {"Display Name": "Gary", "GUID": "profile-guid", "Qualified Name": "Person::gary"}, + ] + assert DummyApp().search_people("a") == [("Tanya", "tanya-guid", "Person::tanya")] + + def test_search_single_result_dict(self, client): + client.find_actor_profiles.return_value = {"Display Name": "Tanya", "GUID": "tanya-guid", + "Qualified Name": "Person::tanya"} + assert DummyApp().search_people("tan") == [("Tanya", "tanya-guid", "Person::tanya")] + + def test_search_error_returns_none(self, client): + client.find_actor_profiles.side_effect = PyegeriaException("boom") + assert DummyApp().search_people("tan") is None + + def test_link_peer(self, client): + assert DummyApp().link_peer("tanya-guid", "Tan", "Met at the summit") is True + owner, peer, body = client.link_peer_person.call_args.args + assert (owner, peer) == ("profile-guid", "tanya-guid") + assert body["properties"]["class"] == "PeerProperties" + assert body["properties"]["peer_relationship_name"] == "Tan" + assert body["properties"]["description"] == "Met at the summit" + + def test_link_peer_error(self, client): + client.link_peer_person.side_effect = PyegeriaException("boom") + app = DummyApp() + assert app.link_peer("tanya-guid") is False + assert app.notices[-1][1] == "error" + + def test_unlink_peer(self, client): + assert DummyApp().unlink_peer("erin-guid") is True + client.unlink_peer_person.assert_called_once_with( + "profile-guid", "erin-guid", {"class": "DeleteRelationshipRequestBody"}) + + def test_unlink_peer_error(self, client): + client.unlink_peer_person.side_effect = PyegeriaException("boom") + app = DummyApp() + assert app.unlink_peer("erin-guid") is False + assert app.notices[-1][1] == "error" + + def test_peer_details(self, client): + client._generate_my_profile_output.return_value = [ + {"Contact Methods": [{"name": "Work email"}], "Communities": [{"name": "Council"}]}] + assert DummyApp().get_peer_details("erin-guid") == ([{"name": "Work email"}], [{"name": "Council"}]) + client.get_actor_profile_by_guid.assert_called_once_with("erin-guid", output_format="JSON") + + def test_peer_details_missing(self, client): + client._generate_my_profile_output.return_value = "No elements found" + app = DummyApp() + assert app.get_peer_details("erin-guid") is None + assert app.notices[-1][1] == "warning" + + +PEER_ROWS = [("Person::erin", "Erin", "Erin Full", "she/her", "erin-guid", "Governance buddy"), + ("Person::peter", "Peter", "Peter Full", "he/him", "peter-guid", "")] + + +class NetworkHostApp(App): + """Host app standing in for MyProfileApp's network methods.""" + + def __init__(self, rows): + super().__init__() + self.rows = list(rows) + self.link_peer = MagicMock(side_effect=self._link) + self.unlink_peer = MagicMock(side_effect=self._unlink) + self.search_people = MagicMock(return_value=[("Tanya", "tanya-guid", "Person::tanya"), + ("Erin", "erin-guid", "Person::erin")]) + self.get_peer_details = MagicMock(return_value=([{"name": "Work email"}], [{"name": "Council"}])) + self.dismissed = None + + def _link(self, guid, nickname="", description=""): + self.rows.append(("Person::new", "New", "", "", guid, description)) + return True + + def _unlink(self, guid): + self.rows = [r for r in self.rows if r[4] != guid] + return True + + def list_my_peers(self): + return list(self.rows) + + async def on_mount(self): + self.screen_under_test = MyNetworkScreen(list(self.rows)) + await self.push_screen(self.screen_under_test, callback=lambda r: setattr(self, "dismissed", r)) + + +class TestMyNetworkScreen: + @pytest.mark.asyncio + async def test_lists_peers(self): + app = NetworkHostApp(PEER_ROWS) + async with app.run_test(): + table = app.screen_under_test.query_one("#my_network_table", DataTable) + assert table.row_count == 2 + assert table.get_row_at(0) == list(PEER_ROWS[0]) + assert app.screen_under_test.query_one("#peer_details_container").display is False + + @pytest.mark.asyncio + async def test_view_details_of_highlighted_peer(self): + app = NetworkHostApp(PEER_ROWS) + async with app.run_test() as pilot: + screen = app.screen_under_test + screen.query_one("#my_network_table", DataTable).move_cursor(row=1) + screen.action_view_details() + await pilot.pause() + app.get_peer_details.assert_called_once_with("peter-guid") + assert screen.query_one("#peer_details_container").display is True + assert screen.query_one("#peer_contacts_table", DataTable).row_count == 1 + assert screen.query_one("#peer_communities_table", DataTable).row_count == 1 + + @pytest.mark.asyncio + async def test_remove_asks_for_confirmation(self): + app = NetworkHostApp(PEER_ROWS) + async with app.run_test() as pilot: + screen = app.screen_under_test + screen.action_remove_peer() + await pilot.pause() + assert isinstance(app.screen, ConfirmRemovePeerScreen) + app.screen.query_one("#cancel_remove_peer_btn", Button).press() + await pilot.pause() + app.unlink_peer.assert_not_called() + assert screen.query_one("#my_network_table", DataTable).row_count == 2 + + @pytest.mark.asyncio + async def test_remove_confirmed_hides_that_peers_details(self): + app = NetworkHostApp(PEER_ROWS) + async with app.run_test() as pilot: + screen = app.screen_under_test + screen.action_view_details() + await pilot.pause() + screen.action_remove_peer() + await pilot.pause() + app.screen.query_one("#confirm_remove_peer_btn", Button).press() + await pilot.pause() + app.unlink_peer.assert_called_once_with("erin-guid") + assert screen.query_one("#my_network_table", DataTable).row_count == 1 + assert screen.query_one("#peer_details_container").display is False + + @pytest.mark.asyncio + async def test_remove_with_empty_network(self): + app = NetworkHostApp([]) + async with app.run_test() as pilot: + app.screen_under_test.action_remove_peer() + await pilot.pause() + assert app.screen is app.screen_under_test + + @pytest.mark.asyncio + async def test_search_and_add_peer(self): + app = NetworkHostApp(PEER_ROWS) + async with app.run_test() as pilot: + screen = app.screen_under_test + search = screen.query_one("#people_search_input", Input) + search.value = "a" + await search.action_submit() + await pilot.pause() + found = screen.query_one("#people_found_table", DataTable) + # Erin is already in the network so isn't offered again + assert found.row_count == 1 + found.focus() + await pilot.press("enter") + await pilot.pause() + screen.query_one("#peer_description_input", Input).value = "Met at the summit" + screen.query_one("#add_peer_btn", Button).press() + await pilot.pause() + app.link_peer.assert_called_once_with("tanya-guid", "", "Met at the summit") + assert screen.query_one("#my_network_table", DataTable).row_count == 3 + assert found.row_count == 0 + + @pytest.mark.asyncio + async def test_quit(self): + app = NetworkHostApp([]) + async with app.run_test() as pilot: + app.screen_under_test.action_quit() + await pilot.pause() + assert app.dismissed == 200 diff --git a/tests/micro-tests/my_profile/test_screens.py b/tests/micro-tests/my_profile/test_screens.py index 80d236b1..b2994cd5 100644 --- a/tests/micro-tests/my_profile/test_screens.py +++ b/tests/micro-tests/my_profile/test_screens.py @@ -776,6 +776,18 @@ async def test_shop_for_data_screen_default_config(self): await pilot.pause() assert app.dismissed_result == [210] + @pytest.mark.parametrize("key", ["b", "escape"]) + @pytest.mark.asyncio + async def test_shop_for_data_screen_go_back_key(self, key): + # Pressed as soon as the screen opens, without focusing a table first: nothing on the + # screen may take focus and swallow the key (a TextArea prompt used to) + app = ScreenTestHostApp(lambda: ShopForDataScreen()) + async with app.run_test() as pilot: + await pilot.pause() + await pilot.press(key) + await pilot.pause() + assert app.dismissed_result == [200] + @pytest.mark.asyncio async def test_shop_for_data_screen_subscribe_action(self): t2 = DataTable(id="digital_product_catalog_table") 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 c3720ae6..a04c3e14 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 @@ -164,6 +164,12 @@ async def test_shop_for_data_callback_exit_codes(self): assert res == 210 assert app.shown_main_screen is True + # go back (b / Escape) dismisses with [200] + app.shown_main_screen = False + res = await app.shop_for_data_callback([200]) + assert res == 200 + assert app.shown_main_screen is True + @pytest.mark.asyncio async def test_shop_for_data_callback_search_for_term(self): app = DummyShopApp() 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 358e90f0..a1da2ac2 100644 --- a/tests/micro-tests/my_profile/test_tech_types_handler.py +++ b/tests/micro-tests/my_profile/test_tech_types_handler.py @@ -119,14 +119,14 @@ async def test_tech_type_callback_valid_selection( ): app = DummyTechTypesApp(backend) if backend.live: - backend.patch("tech_types_handler.AutomatedCuration") + backend.patch("tech_types_handler.Egeria") # 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) + backend.always_fake("tech_types_handler.Egeria", returns=mock_ac) selection = "tech-type-guid-999" res = await app.tech_type_callback(selection) @@ -161,12 +161,43 @@ async def test_tech_type_callback_exception(self, backend): app = DummyTechTypesApp(backend) mock_ac = MagicMock() mock_ac.create_egeria_bearer_token.side_effect = PyegeriaException("Auth failed") - backend.always_fake("tech_types_handler.AutomatedCuration", returns=mock_ac) + backend.always_fake("tech_types_handler.Egeria", returns=mock_ac) - await app.tech_type_callback("tech-type-guid-999") - assert len(app.pushed_screens) == 2 # StatusScreen pushed on error, then TechnologyTypeOptionsScreen + res = await app.tech_type_callback("tech-type-guid-999") + # Only the StatusScreen: the options screen is not opened after a failed lookup + assert res == 417 + assert len(app.pushed_screens) == 1 assert isinstance(app.pushed_screens[0][0], StatusScreen) + @pytest.mark.parametrize("result", ["201", 201, "200", 200]) + @pytest.mark.asyncio + async def test_tech_type_callback_go_back(self, backend, result): + # TechnologyTypesScreen dismisses go back/quit as the strings "201"/"200" + app = DummyTechTypesApp(backend) + mock_egeria = MagicMock() + backend.always_fake("tech_types_handler.Egeria", returns=mock_egeria) + + res = await app.tech_type_callback(result) + assert res == result + assert app.shown_main_screen is True + assert app.pushed_screens == [] + mock_egeria.get_tech_type_detail.assert_not_called() + + @pytest.mark.asyncio + async def test_tech_type_callback_no_details(self, backend): + # Always faked: Egeria answers with a string when the type has no details + app = DummyTechTypesApp(backend) + app.notify = MagicMock() + mock_egeria = MagicMock() + mock_egeria.get_tech_type_detail.return_value = "No elements found" + backend.always_fake("tech_types_handler.Egeria", returns=mock_egeria) + + res = await app.tech_type_callback("Unknown Type") + assert res == 200 + assert app.shown_main_screen is True + assert app.pushed_screens == [] + app.notify.assert_called_once() + @pytest.mark.asyncio async def test_tech_type_options_callback_template(self): app = DummyTechTypesApp()