diff --git a/docs/My-Egeria-Doc.md b/docs/My-Egeria-Doc.md index 25584095..6b6adf93 100644 --- a/docs/My-Egeria-Doc.md +++ b/docs/My-Egeria-Doc.md @@ -31,6 +31,8 @@ All source files named here are in `my_egeria/my_egeria/DemoCode/My_Profile/`. | `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` | +| `community_handler.py`, `CommunitiesScreen.py` | `CommunitiesMixin` (community membership through `AssignmentScope` relationships), `CommunitiesScreen` and `ConfirmLeaveCommunityScreen` | +| `CommunityScreen.py` | `CommunityScreen` (one community's members, blog and comments), `NewBlogPostScreen` and `AddCommunityCommentScreen` | | `UserIdentitiesScreen.py` | Read-only user identity table | | `StatusScreen.py` | Result message with copy-GUID | | `profile_utils.py` | Environment check, Egeria access check (`check_egeria_access`) and data clean-up helpers | @@ -77,6 +79,15 @@ classDiagram link_peer() / unlink_peer() get_peer_details() } + class CommunitiesMixin { + show_communities() + list_my_communities() + search_communities() + get_community_members() + join_community() / leave_community() + get_community_blog_posts() / add_community_blog_post() + get_community_comments() / add_community_comment() + } App <|-- MyProfileApp TechTypesMixin <|-- MyProfileApp ShopForDataMixin <|-- MyProfileApp @@ -85,6 +96,7 @@ classDiagram FeedbackMixin <|-- MyProfileApp BookmarksMixin <|-- MyProfileApp NetworkMixin <|-- MyProfileApp + CommunitiesMixin <|-- MyProfileApp ``` - **App bindings:** @@ -183,6 +195,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()` | +| Work with Communities | `show_communities()` | | Change User | `ChangeUserScreen()`, callback `change_user_callback` | | Leave Feedback | `action_feedback()` | | Feedback Log | `action_view_feedback_log()` (owner only) | @@ -256,7 +269,7 @@ The screen stays open for repeated adds and dismisses with `200`. `BaseAddScreen.validate(values)` lets a subclass reject values before any Egeria call. To-Do checks that its priority is a whole number; Project checks its kind and its `YYYY-MM-DD` dates. -Team membership in Egeria is a `TeamMember` role (a PersonRole subtype) appointed to the person and scoped to the team by an `AssignmentScope` relationship. This is the same model Dr.Egeria's `Link Team Membership` uses. `link_element()` can read the form values through `self.last_values`. Communities have no "link to me" step, because pyegeria has no community-membership call. +Team membership in Egeria is a `TeamMember` role (a PersonRole subtype) appointed to the person and scoped to the team by an `AssignmentScope` relationship. This is the same model Dr.Egeria's `Link Team Membership` uses. `link_element()` can read the form values through `self.last_values`. Communities have no "link to me" step; the user joins a community separately from [Work with Communities](#work-with-communities). When any Add screen closes, its `add_*_callback` calls `_after_add()`. That pops back to the main screen, which drops the edit screen's stale rows, and schedules `action_refresh()` so the new element appears. @@ -564,6 +577,98 @@ sequenceDiagram --- +## Work with Communities + +A user belongs to a community when an actor is linked to it by an `AssignmentScope` relationship, whose `assignmentType` says how they take part. The actor is either the user's own profile or a role they are appointed to. A community's blog is a `NoteLog` attached to it holding `BlogEntry` notes, and comments are attached to the community directly. `CommunitiesMixin` (`community_handler.py`) makes the Egeria calls; `CommunitiesScreen` and `CommunityScreen` only drive it through `self.app`. Each call builds a fresh `Egeria` client with a bearer token (`_community_client()`) and closes its session afterwards. + +```mermaid +sequenceDiagram + autonumber + actor User + participant App as CommunitiesMixin + participant CS as CommunitiesScreen + participant Confirm as ConfirmLeaveCommunityScreen + participant EG as Egeria + + User->>App: Other Functions → Work with Communities + App->>EG: get_my_profile(JSON) + App->>CS: push(my community rows) + CS->>App: search_communities("*") + App->>EG: find_communities(JSON, graph_query_depth=0) + opt open a community (Enter / ctrl+o) — see the next diagram + CS->>CS: push CommunityScreen(GUID, name) + end + opt join (ctrl+n) + CS->>App: join_community(GUID, assignment type) + App->>EG: link_assignment_scope(GUID, user_GUID, AssignmentScopeProperties) + CS->>App: list_my_communities() and search again + end + opt leave (ctrl+d) + CS->>Confirm: push(community name) + Confirm-->>CS: True / False + CS->>App: leave_community(GUID) + App->>EG: detach_assignment_scope(GUID, user_GUID) + CS->>App: list_my_communities() and search again + end +``` + +| Function | What it does | +| :--- | :--- | +| `show_communities()` | Refuses (with a notice) if `user_GUID` isn't set yet; otherwise pushes `CommunitiesScreen(list_my_communities())` | +| `list_my_communities()` | `get_my_profile(JSON)`, then reads the profile's own `assignmentScope` links and each `performsRoles` appointment's `nestedElements`, keeping only `Community` elements. Returns one `(name, assignment types, linked through, description, GUID)` row per community; "linked through" is `Your profile` and/or `Role: `. `None` if Egeria can't be reached | +| `search_communities(search_string)` | `find_communities(..., ignore_case=True, graph_query_depth=0)` → `(name, description, qualified name, GUID)` rows. The screen drops communities the user already belongs to | +| `join_community(guid, assignment_type)` | `link_assignment_scope(guid, user_GUID, NewRelationshipRequestBody)` with `AssignmentScopeProperties`. The screen offers only `Contributor` and `Observer` | +| `leave_community(guid)` | `detach_assignment_scope(guid, user_GUID, DeleteRelationshipRequestBody)`, using the same GUID order as `link_assignment_scope` (scope first, actor second) | + +- **Leaving only unlinks the profile.** The screen refuses to leave a community reached only through a role, because the role may be shared and belongs to whoever manages it. +- **Refresh.** After a successful join or leave the screen re-reads the user's communities and repeats the last search. +- **Errors.** Every Egeria failure is reported with `notify(...)` and the screen stays open; none of them exit the app. +- **Return code.** `CommunitiesScreen` dismisses with `200` on `q`; `communities_callback` just logs it. + +### A community: members, blog and comments + +`CommunityScreen(guid, name)` loads the three lists when it mounts and reloads a list after a successful add. `ctrl+n` opens `AddCommunityCommentScreen` when the comments table has focus, otherwise `NewBlogPostScreen`; each form dismisses with a dict, or `None` on cancel. + +```mermaid +sequenceDiagram + autonumber + actor User + participant CoS as CommunityScreen + participant App as CommunitiesMixin + participant EG as Egeria + + CoS->>App: get_community_members(GUID) + App->>EG: get_community_by_guid(JSON, graph_query_depth=1) + CoS->>App: get_community_blog_posts(GUID) + App->>EG: get_attached_note_logs(GUID) + App->>EG: get_notes_for_note_log(log, "BlogEntry") for each log + CoS->>App: get_community_comments(GUID) + App->>EG: get_attached_comments(GUID) + opt new blog post (NewBlogPostScreen) + CoS->>App: add_community_blog_post(GUID, name, title, text) + App->>EG: get_attached_note_logs(GUID) + App->>EG: create_note_log(GUID, NewAttachmentRequestBody) if the blog doesn't exist + App->>EG: create_note(log, BlogEntryProperties) + end + opt add comment (AddCommunityCommentScreen) + CoS->>App: add_community_comment(GUID, text, type) + App->>EG: add_comment_to_element(GUID, text, type) + end +``` + +| Function | What it does | +| :--- | :--- | +| `get_community_members(guid)` | `get_community_by_guid(guid, JSON, graph_query_depth=1)`, reading `assignedActors` → `(name, actor type, assignment type, qualified name, GUID)` rows | +| `get_community_blog_posts(guid)` | Reads the `BlogEntry` notes from every note log attached to the community, so posts written by other tools are included and the notifications some sample communities' logs hold are not. Returns the `LATEST_POSTS` (20) newest as `(title, author, posted, text, GUID)` rows; author and time come from the element header's `versions` | +| `add_community_blog_post(guid, name, title, text)` | Finds the note log with qualified name `NoteLog::::CommunityBlog` among those attached to the community, creating it (attached, display name "_name_ Blog") on the first post, then `create_note(log, BlogEntryProperties)` | +| `get_community_comments(guid)` | `get_attached_comments(guid, graph_query_depth=0)` → `(comment, type, author, posted, GUID)` rows, newest first | +| `add_community_comment(guid, text, type)` | `add_comment_to_element(guid, text, type)`; the form offers pyegeria's `COMMENT_TYPES` | + +- **Return codes.** `CommunityScreen` dismisses with `200` on `Escape` (back to the communities list) and `210` on `q`; on `210` `CommunitiesScreen` dismisses itself too, so `q` reaches the main screen. +- **Membership isn't checked.** Anyone can read and add to a community's blog and comments; Egeria's quickstart deployment enforces no access rules here either. + +--- + ## Status Screen `StatusScreen(message)` shows a read-only message: @@ -593,7 +698,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; My Network; feedback note logs; tabular data; `update_actor_profile`; `find_collections` | +| `Egeria` | Element create, link and delete; comments from main; bookmarks; My Network; communities; feedback note logs; tabular data; `update_actor_profile`; `find_collections` | | `EgeriaTech` | Comments screen | | `AutomatedCuration` | Technology types | | `ProductManager` | Digital subscriptions | @@ -634,6 +739,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. +**Work with Communities.** `test_community_handler.py` covers `CommunitiesMixin`, `CommunitiesScreen` and `CommunityScreen`, including memberships through roles, the leave confirmation, refusing to leave a role membership, creating the blog on the first post and reusing it afterwards, and the post and comment forms; `test_my_egeria.py` checks the menu routing. The blog and comment functions have also been run against a live server (2026-10-28); the screens themselves have not been tried by hand there yet. + **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. @@ -646,6 +753,7 @@ The user-visible limitations are listed in the [user manual](my_egeria_manual.md - `get_data_product_catalog_table`, `get_guid_for_qualified_name`, `display_glossary_term_details`, `tech_type_processes_details`, `unpack_egeria_data` and `display_selected_data_specification` are defined but never called. - Bookmarks, My Collections, team joining, subscription links and starting processes have only been tested against fakes; they haven't yet been run against a live Egeria server. +- Work with Communities: the blog and comment functions have been run against a live server, but `list_my_communities`, `search_communities`, `get_community_members`, `join_community`, `leave_community` and the screens have only been tested against fakes. --- License: [CC BY 4.0](https://creativecommons.org/licenses/by/4.0/), diff --git a/docs/my_egeria_manual.md b/docs/my_egeria_manual.md index bbd969c9..3f11635d 100644 --- a/docs/my_egeria_manual.md +++ b/docs/my_egeria_manual.md @@ -289,6 +289,26 @@ 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. +### Work with Communities +The screen lists the communities you belong to (name, assignment type, what links you to it, description and GUID), lets you find others to join, and opens a community to show its members, blog and comments. If your profile has not finished loading, the app tells you to wait and does not open the screen. + +- **Linked Through** is **Your profile** if you joined the community yourself, or **Role: _name_** if you belong through a role you hold. A community you reach both ways is listed once. +- **Open a community:** highlight a community in either list and press `Enter`, `ctrl+o` or **Open Community**. See [A community](#a-community) below. +- **Find and join:** type a search term in the search field (`*` for all) and press `Enter`. Communities you already belong to are left out. Choose **Contributor** or **Observer**, highlight a community and press `ctrl+n` (or **Join Community**). Leaders and administrators are assigned by the people who manage the community, not chosen here. +- **Leave:** highlight one of your communities and press `ctrl+d` (or **Leave Community**), then confirm with **Leave**. **Cancel** or `Escape` keeps you in it. Only the link between you and the community is removed. You can't leave a community you belong to through a role; ask whoever manages that role. +- `Escape` or `q` returns to the main screen. + +Membership is an `AssignmentScope` relationship from your profile (or a role) to the community, and the assignment type is stored on that relationship. + +#### A community +Opening a community shows three lists: its **latest blog posts** (the 20 newest, with title, author and date; the highlighted post's text is shown underneath), its **members** (people or roles) and the **comments** on it, newest first. You don't have to be a member to read or add to them. + +- **New blog post:** press `ctrl+n` (or **New Blog Post**), enter a title and the text, then press `ctrl+s` (or **Post**). **Cancel** or `Escape` discards it. +- **Add a comment:** select the comments list and press `ctrl+n` (or press **Add Comment**), choose a comment type (Standard Comment, Question, Answer, Suggestion, Usage Experience, Requirement or Other), enter the comment and press `ctrl+s` (or **Add Comment**). +- `Escape` returns to the communities list; `q` returns to the main screen. + +The community's blog is a note log attached to the community, created when the first post is made; each post is a `BlogEntry` in it. Comments are attached to the community itself. + ### Change User Log in as a different Egeria user and reload the app for them. See [The Change User screen](#the-change-user-screen). @@ -300,7 +320,7 @@ See [Leaving Feedback](#leaving-feedback). Every screen follows the same rules: - `Escape` goes back one level, on every screen except the main screen. -- `q` returns to the main screen, or quits the app when you are on the main screen. Screens that are mostly text fields (Add screens, Add Comment, Create Subscription, Create/Edit Profile, Search for Term, Feedback, Template and Process forms) have no `q`, because it would be typed into the field; use `Escape`. +- `q` returns to the main screen, or quits the app when you are on the main screen. Screens that are mostly text fields (Add screens, Add Comment, Create Subscription, Create/Edit Profile, Search for Term, Feedback, New Blog Post, Template and Process forms) have no `q`, because it would be typed into the field; use `Escape`. - Every other action is a `ctrl` key, and a key means the same thing wherever it appears. `ctrl` keys work even while a text field has focus. | Key | Action | Where | @@ -309,10 +329,10 @@ Every screen follows the same rules: | `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 | -| `ctrl+n` | Add a new item | Edit screens (row), Bookmarks, Comments, Main screen (comment) | -| `ctrl+d` | Delete / remove the highlighted item | Edit screens, Bookmarks, My Network | -| `ctrl+o` | Open the highlighted item | Main screen (comments), Comments (responses), My Network (person details) | +| `ctrl+s` | Save / submit the form | Add screens, Add Comment, Create Subscription, Create/Edit Profile, Feedback, New Blog Post | +| `ctrl+n` | Add a new item | Edit screens (row), Bookmarks, Comments, Main screen (comment), Communities (join), a community (blog post or comment) | +| `ctrl+d` | Delete / remove the highlighted item | Edit screens, Bookmarks, My Network, Communities (leave) | +| `ctrl+o` | Open the highlighted item | Main screen (comments), Comments (responses), My Network (person details), Communities (open a community) | | `ctrl+t` | Edit the selected table | Main screen | | `ctrl+k` | Bookmark the highlighted row | Main screen, Shop for Data | | `ctrl+b` | Bookmarks, with the highlighted row's GUID ready to add | Main screen, Shop for Data | @@ -359,16 +379,17 @@ The full list is in `RETURN_CODES.md` in the app folder. - 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. 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. +**Something I added doesn't appear.** Press `ctrl+r` on the main screen. Communities you create are not linked to you (see Known Limitations), so they don't appear in User Associations until you join them from **Work with Communities**. **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 -As of 2026-10-01, these features are incomplete or don't work as described elsewhere: +As of 2026-10-28, these features are incomplete or don't work as described elsewhere: -- **Communities you add** are not linked to you. pyegeria has no call for community membership, so the community is created but won't appear in User Associations. +- **Communities you add** are not linked to you automatically. The community is created but won't appear in User Associations until you join it from **Work with Communities**. - **Not yet verified against a live Egeria server:** bookmarks, My Collections, joining a team, subscription links and starting processes. The test server was unavailable when they were added, so they have only been tested against fakes. +- **Work with Communities** has only partly been tried against a live server. Reading and adding blog posts and comments works there; the rest (listing, finding, joining and leaving communities, and the screens themselves) has so far only been tested against fakes. ## Running the Tests diff --git a/my_egeria/my_egeria/DemoCode/My_Profile/CommunitiesScreen.py b/my_egeria/my_egeria/DemoCode/My_Profile/CommunitiesScreen.py new file mode 100644 index 00000000..0edfb2ad --- /dev/null +++ b/my_egeria/my_egeria/DemoCode/My_Profile/CommunitiesScreen.py @@ -0,0 +1,273 @@ +""" + PDX-License-Identifier: Apache-2.0 + Copyright Contributors to the ODPi Egeria project. + + This file provides the Work with Communities screen for My Profile. + +""" +from textual import on +from textual.app import ComposeResult +from textual.containers import ScrollableContainer, Horizontal +from textual.screen import ModalScreen +from textual.widgets import DataTable, Header, Static, Footer, Input, Button, Select + +from CommunityScreen import CommunityScreen + +# The "Linked Through" value of a membership held by the user's own profile, which they can leave +DIRECT_LINK = "Your profile" +# Assignment types a user can give themselves when joining. Leaders and administrators are +# assigned by the people who manage the community (https://egeria-project.org/concepts/community/) +JOIN_ASSIGNMENT_TYPES = ["Contributor", "Observer"] + + +def is_direct_membership(row) -> bool: + """True if a my-communities row includes a link from the user's own profile.""" + return DIRECT_LINK in row[2].split("; ") + + +class ConfirmLeaveCommunityScreen(ModalScreen[bool]): + """Confirmation prompt shown before the user leaves a community. + + Dismisses with True to proceed with leaving, False to cancel. + """ + + BINDINGS = [("escape", "cancel", "Cancel")] + + CSS_PATH = "my_profile.tcss" + + def __init__(self, community_name: str, *args, **kwargs): + super().__init__(*args, **kwargs, id="confirm_leave_community_screen") + self.community_name = community_name + + def compose(self) -> ComposeResult: + yield Header(show_clock=True) + yield Static("Confirm Leaving", classes="span-3", id="confirm_leave_community_title") + yield ScrollableContainer( + Static(f"Leave {self.community_name}?"), + Static("The community is not changed, only the link between you and it."), + Horizontal( + Button("Leave", id="confirm_leave_community_btn", variant="error"), + Button("Cancel", id="cancel_leave_community_btn", variant="primary"), + ), + id="confirm_leave_community_container", + ) + yield Footer() + + def on_mount(self) -> None: + self.title = "Egeria - my_profile" + self.sub_title = "Confirm Leaving - Communities" + + @on(Button.Pressed, "#confirm_leave_community_btn") + def handle_confirm_button(self, event: Button.Pressed) -> None: + self.dismiss(True) + + @on(Button.Pressed, "#cancel_leave_community_btn") + def handle_cancel_button(self, event: Button.Pressed) -> None: + self.dismiss(False) + + def action_cancel(self) -> None: + self.dismiss(False) + + +class CommunitiesScreen(ModalScreen): + """List the communities the current user belongs to, find others to join, open a + community (members, blog and comments) and leave a community. + + The Egeria work is done by the app's CommunitiesMixin (list_my_communities, + search_communities, join_community, leave_community); this screen only drives it. + """ + + BINDINGS = [ + ("escape", "quit", "Back"), + ("q", "quit", "Main Menu"), + ("ctrl+o", "open_community", "Open Community"), + ("ctrl+n", "join_community", "Join Community"), + ("ctrl+d", "leave_community", "Leave Community"), + ] + + CSS_PATH = "my_profile.tcss" + + def __init__(self, my_communities, *args, **kwargs): + super().__init__(id="communities_screen", *args, **kwargs) + self.title = "Egeria" + self.sub_title = "Communities" + # Rows of (name, assignment type, linked through, description, GUID) + self.my_communities_data = list(my_communities or []) + # The last search, repeated after joining or leaving so the found list stays current + self.search_string = "*" + + def compose(self) -> ComposeResult: + yield Header(show_clock=True) + yield Static("Communities", classes="span-3", id="communities_title") + yield ScrollableContainer( + Static("The communities you belong to. Enter or ctrl+o opens one (members, blog and comments); " + "ctrl+d leaves."), + DataTable(id="my_communities_table", zebra_stripes=True, cursor_type="row"), + Horizontal( + Button("Open Community", id="open_my_community_btn", variant="primary"), + Button("Leave Community", id="leave_community_btn", variant="error"), + id="my_communities_buttons", + ), + id="my_communities_container", + ) + yield ScrollableContainer( + Static("Find other communities: enter a search term ('*' for all) and press return"), + Input(value=self.search_string, placeholder="Search term, or * for all", id="community_search_input"), + DataTable(id="communities_found_table", zebra_stripes=True, cursor_type="row"), + Horizontal( + Select([(t, t) for t in JOIN_ASSIGNMENT_TYPES], value=JOIN_ASSIGNMENT_TYPES[0], + allow_blank=False, id="join_assignment_type"), + Button("Join Community", id="join_community_btn", variant="primary"), + Button("Open Community", id="open_found_community_btn", variant="default"), + id="found_communities_buttons", + ), + id="find_communities_container", + ) + yield Footer() + + def on_mount(self) -> None: + self.query_one("#my_communities_table", DataTable).add_columns( + "Name", "Assignment Type", "Linked Through", "Description", "GUID") + self.query_one("#communities_found_table", DataTable).add_columns( + "Name", "Description", "Qualified Name", "GUID") + self._fill_my_table() + self._search() + + def _fill_my_table(self) -> None: + table = self.query_one("#my_communities_table", DataTable) + table.clear() + for row in self.my_communities_data: + table.add_row(*row, key=row[4]) + if not self.my_communities_data: + self.notify("You don't belong to any communities yet - find one to join", timeout=5) + + def _search(self) -> None: + """Run the current search, listing only communities the user doesn't already belong to.""" + found = self.app.search_communities(self.search_string) + if found is None: + return + table = self.query_one("#communities_found_table", DataTable) + table.clear() + mine = {row[4] for row in self.my_communities_data} + for row in found: + if row[3] not in mine: + table.add_row(*row, key=row[3]) + if table.row_count == 0: + self.notify(f"No other communities found matching {self.search_string}", timeout=5, + severity="warning") + + def _reload(self) -> None: + rows = self.app.list_my_communities() + if rows is not None: + self.my_communities_data = rows + self._fill_my_table() + self._search() + + @staticmethod + def _highlighted(table: DataTable) -> list | None: + if table.row_count == 0: + return None + return table.get_row_at(table.cursor_row) + + def action_quit(self) -> None: + """ The user elects to quit the communities function """ + self.dismiss(200) + + # --- opening a community ---------------------------------------------- + + def _open(self, community_guid: str, community_name: str) -> None: + def on_close(result) -> None: + # q on the community screen goes on to the main screen + if result == 210: + self.dismiss(200) + + self.app.push_screen(CommunityScreen(community_guid, community_name), callback=on_close) + + def action_open_community(self) -> None: + """ Open the highlighted community in whichever list has focus """ + found_table = self.query_one("#communities_found_table", DataTable) + if self.focused is found_table: + self._open_found() + else: + self._open_mine() + + def _open_mine(self) -> None: + row = self._highlighted(self.query_one("#my_communities_table", DataTable)) + if row is None: + self.notify("You don't belong to any communities to select", timeout=5, severity="warning") + return + self._open(row[4], row[0]) + + def _open_found(self) -> None: + row = self._highlighted(self.query_one("#communities_found_table", DataTable)) + if row is None: + self.notify("There are no communities found to select", timeout=5, severity="warning") + return + self._open(row[3], row[0]) + + @on(Button.Pressed, "#open_my_community_btn") + def handle_open_mine_button(self, event: Button.Pressed) -> None: + self._open_mine() + + @on(Button.Pressed, "#open_found_community_btn") + def handle_open_found_button(self, event: Button.Pressed) -> None: + self._open_found() + + @on(DataTable.RowSelected, "#my_communities_table") + def handle_my_community_selected(self, event: DataTable.RowSelected) -> None: + self._open_mine() + + @on(DataTable.RowSelected, "#communities_found_table") + def handle_found_community_selected(self, event: DataTable.RowSelected) -> None: + self._open_found() + + # --- finding and joining ------------------------------------------------- + + @on(Input.Submitted, "#community_search_input") + def handle_community_search(self, event: Input.Submitted) -> None: + self.search_string = event.value.strip() or "*" + self._search() + found_table = self.query_one("#communities_found_table", DataTable) + if found_table.row_count: + found_table.focus() + + @on(Button.Pressed, "#join_community_btn") + def handle_join_button(self, event: Button.Pressed) -> None: + self.action_join_community() + + def action_join_community(self) -> None: + """ The user joins the highlighted found community, with the chosen assignment type """ + row = self._highlighted(self.query_one("#communities_found_table", DataTable)) + if row is None: + self.notify("Find a community to join first", timeout=5, severity="warning") + return + assignment_type = self.query_one("#join_assignment_type", Select).value + if self.app.join_community(row[3], assignment_type): + self._reload() + + # --- leaving ------------------------------------------------------------- + + @on(Button.Pressed, "#leave_community_btn") + def handle_leave_button(self, event: Button.Pressed) -> None: + self.action_leave_community() + + def action_leave_community(self) -> None: + """ The user wants to leave the highlighted community, after confirmation """ + row = self._highlighted(self.query_one("#my_communities_table", DataTable)) + if row is None: + self.notify("You don't belong to any communities to leave", timeout=5, severity="warning") + return + community_name, community_guid = row[0], row[4] + if not is_direct_membership(row): + self.notify(f"You belong to {community_name} through a role ({row[2]}), so you can't leave it " + f"here; ask whoever manages that role", timeout=10, severity="warning") + return + + def on_confirm(confirmed: bool | None) -> None: + if not confirmed: + self.notify(f"You are still a member of {community_name}", timeout=5) + return + if self.app.leave_community(community_guid): + self._reload() + + self.app.push_screen(ConfirmLeaveCommunityScreen(community_name), callback=on_confirm) diff --git a/my_egeria/my_egeria/DemoCode/My_Profile/CommunityScreen.py b/my_egeria/my_egeria/DemoCode/My_Profile/CommunityScreen.py new file mode 100644 index 00000000..6dc8f679 --- /dev/null +++ b/my_egeria/my_egeria/DemoCode/My_Profile/CommunityScreen.py @@ -0,0 +1,290 @@ +""" + PDX-License-Identifier: Apache-2.0 + Copyright Contributors to the ODPi Egeria project. + + This file provides the screen for a single community in My Profile: its members, + the latest posts on its blog and the comments attached to it, plus the forms to + add a blog post or a comment. + +""" +from textual import on +from textual.app import ComposeResult +from textual.containers import ScrollableContainer, Horizontal +from textual.screen import ModalScreen +from textual.widgets import DataTable, Header, Static, Footer, Input, Button, Select, TextArea + +from pyegeria import COMMENT_TYPES + + +def comment_type_label(comment_type: str) -> str: + """'USAGE_EXPERIENCE' -> 'Usage Experience'""" + return comment_type.replace("_", " ").title() + + +class NewBlogPostScreen(ModalScreen): + """Collect the title and text of a new post for a community's blog. + + Dismisses with {"title", "text"}, or None if cancelled. + """ + + BINDINGS = [ + ("escape", "cancel", "Back"), + ("ctrl+s", "submit", "Post"), + ] + + CSS_PATH = "my_profile.tcss" + + def __init__(self, community_name: str, *args, **kwargs): + super().__init__(*args, **kwargs, id="new_blog_post_screen") + self.community_name = community_name + + def compose(self) -> ComposeResult: + yield Header(show_clock=True) + yield Static("New Blog Post", classes="span-3", id="new_blog_post_title") + yield ScrollableContainer( + Static(f"[b]Post to the {self.community_name} blog[/b]"), + Static("Title"), + Input(placeholder="Title of the post", id="blog_post_title_input"), + Static("Post"), + TextArea(id="blog_post_text"), + Horizontal( + Button("Post", id="submit_blog_post_btn", variant="primary"), + Button("Cancel", id="cancel_blog_post_btn", variant="warning"), + ), + id="new_blog_post_container", + ) + yield Footer() + + def on_mount(self) -> None: + self.title = "Egeria - my_profile" + self.sub_title = "New Blog Post - Communities" + self.query_one("#blog_post_title_input", Input).focus() + + def action_submit(self) -> None: + title = self.query_one("#blog_post_title_input", Input).value.strip() + text = self.query_one("#blog_post_text", TextArea).text.strip() + if not title or not text: + self.notify("Please enter both a title and the text of the post", timeout=5, severity="warning") + return + self.dismiss({"title": title, "text": text}) + + def action_cancel(self) -> None: + self.dismiss(None) + + @on(Button.Pressed, "#submit_blog_post_btn") + def handle_submit_button(self, event: Button.Pressed) -> None: + self.action_submit() + + @on(Button.Pressed, "#cancel_blog_post_btn") + def handle_cancel_button(self, event: Button.Pressed) -> None: + self.action_cancel() + + +class AddCommunityCommentScreen(ModalScreen): + """Collect the text and type of a comment on a community. + + Dismisses with {"text", "comment_type"}, or None if cancelled. + """ + + BINDINGS = [ + ("escape", "cancel", "Back"), + ("ctrl+s", "submit", "Add Comment"), + ] + + CSS_PATH = "my_profile.tcss" + + def __init__(self, community_name: str, *args, **kwargs): + super().__init__(*args, **kwargs, id="add_community_comment_screen") + self.community_name = community_name + + def compose(self) -> ComposeResult: + yield Header(show_clock=True) + yield Static("Add Comment", classes="span-3", id="add_community_comment_title") + yield ScrollableContainer( + Static(f"[b]Comment on {self.community_name}[/b]"), + Static("Comment type"), + Select([(comment_type_label(t), t) for t in COMMENT_TYPES], value=COMMENT_TYPES[0], + allow_blank=False, id="community_comment_type"), + Static("Comment"), + TextArea(id="community_comment_text"), + Horizontal( + Button("Add Comment", id="submit_community_comment_btn", variant="primary"), + Button("Cancel", id="cancel_community_comment_btn", variant="warning"), + ), + id="add_community_comment_container", + ) + yield Footer() + + def on_mount(self) -> None: + self.title = "Egeria - my_profile" + self.sub_title = "Add Comment - Communities" + self.query_one("#community_comment_text", TextArea).focus() + + def action_submit(self) -> None: + text = self.query_one("#community_comment_text", TextArea).text.strip() + if not text: + self.notify("Please enter a comment", timeout=5, severity="warning") + return + self.dismiss({"text": text, "comment_type": self.query_one("#community_comment_type", Select).value}) + + def action_cancel(self) -> None: + self.dismiss(None) + + @on(Button.Pressed, "#submit_community_comment_btn") + def handle_submit_button(self, event: Button.Pressed) -> None: + self.action_submit() + + @on(Button.Pressed, "#cancel_community_comment_btn") + def handle_cancel_button(self, event: Button.Pressed) -> None: + self.action_cancel() + + +class CommunityScreen(ModalScreen): + """One community: its members, the latest posts on its blog and its comments. + + The Egeria work is done by the app's CommunitiesMixin (get_community_members, + get_community_blog_posts, add_community_blog_post, get_community_comments, + add_community_comment); this screen only drives it. + + Dismisses with 200 to go back to the communities list, or 210 to go on to the main screen. + """ + + BINDINGS = [ + ("escape", "back", "Back"), + ("q", "quit", "Main Menu"), + ("ctrl+n", "add_item", "New Post / Comment"), + ] + + CSS_PATH = "my_profile.tcss" + + def __init__(self, community_guid: str, community_name: str, *args, **kwargs): + super().__init__(id="community_screen", *args, **kwargs) + self.title = "Egeria" + self.sub_title = f"Community - {community_name}" + self.community_guid = community_guid + self.community_name = community_name + # Full text of each blog post, by GUID, shown below the posts table + self.post_text: dict[str, str] = {} + + def compose(self) -> ComposeResult: + yield Header(show_clock=True) + yield Static(self.community_name, classes="span-3", id="community_title") + yield ScrollableContainer( + Static("[b]Latest blog posts[/b] [dim]ctrl+n adds a post[/dim]"), + DataTable(id="community_posts_table", zebra_stripes=True, cursor_type="row"), + Static("", id="community_post_text"), + Button("New Blog Post", id="new_blog_post_btn", variant="primary"), + id="community_blog_container", + ) + yield ScrollableContainer( + Static("[b]Members[/b] [dim](an actor may be a person or a role)[/dim]"), + DataTable(id="community_members_table", zebra_stripes=True, cursor_type="row"), + id="community_members_container", + ) + yield ScrollableContainer( + Static("[b]Comments[/b] [dim]with this list selected, ctrl+n adds a comment[/dim]"), + DataTable(id="community_comments_table", zebra_stripes=True, cursor_type="row"), + Button("Add Comment", id="add_community_comment_btn", variant="primary"), + id="community_comments_container", + ) + yield Footer() + + def on_mount(self) -> None: + self.query_one("#community_posts_table", DataTable).add_columns("Title", "Author", "Posted", "GUID") + self.query_one("#community_members_table", DataTable).add_columns( + "Name", "Actor Type", "Assignment Type", "Qualified Name", "GUID") + self.query_one("#community_comments_table", DataTable).add_columns( + "Comment", "Type", "Author", "Posted", "GUID") + self._load_members() + self._load_posts() + self._load_comments() + + def _load_members(self) -> None: + members = self.app.get_community_members(self.community_guid) + if members is None: + return + table = self.query_one("#community_members_table", DataTable) + table.clear() + for row in members: + table.add_row(*row) + + def _load_posts(self) -> None: + """Rows of (title, author, posted, text, GUID), newest first.""" + posts = self.app.get_community_blog_posts(self.community_guid) + if posts is None: + return + table = self.query_one("#community_posts_table", DataTable) + table.clear() + self.post_text = {} + for title, author, posted, text, guid in posts: + table.add_row(title, author, posted, guid, key=guid) + self.post_text[guid] = text + self._show_post_text() + + def _load_comments(self) -> None: + comments = self.app.get_community_comments(self.community_guid) + if comments is None: + return + table = self.query_one("#community_comments_table", DataTable) + table.clear() + for row in comments: + table.add_row(*row, key=row[4]) + + def _show_post_text(self) -> None: + table = self.query_one("#community_posts_table", DataTable) + if table.row_count: + guid = table.get_row_at(table.cursor_row)[3] + text = self.post_text.get(guid, "") + else: + text = "[dim]No posts yet - be the first to post[/dim]" + self.query_one("#community_post_text", Static).update(text) + + @on(DataTable.RowHighlighted, "#community_posts_table") + def handle_post_highlighted(self, event: DataTable.RowHighlighted) -> None: + self._show_post_text() + + def action_back(self) -> None: + """ Back to the communities list """ + self.dismiss(200) + + def action_quit(self) -> None: + """ Back to the main screen """ + self.dismiss(210) + + def action_add_item(self) -> None: + """ Add a comment if the comments list has focus, otherwise a blog post """ + if self.focused is self.query_one("#community_comments_table", DataTable): + self.action_add_comment() + else: + self.action_new_post() + + # --- blog ---------------------------------------------------------------- + + @on(Button.Pressed, "#new_blog_post_btn") + def handle_new_post_button(self, event: Button.Pressed) -> None: + self.action_new_post() + + def action_new_post(self) -> None: + def on_post(post: dict | None) -> None: + if not post: + return + if self.app.add_community_blog_post(self.community_guid, self.community_name, + post["title"], post["text"]): + self._load_posts() + + self.app.push_screen(NewBlogPostScreen(self.community_name), callback=on_post) + + # --- comments ------------------------------------------------------------ + + @on(Button.Pressed, "#add_community_comment_btn") + def handle_add_comment_button(self, event: Button.Pressed) -> None: + self.action_add_comment() + + def action_add_comment(self) -> None: + def on_comment(comment: dict | None) -> None: + if not comment: + return + if self.app.add_community_comment(self.community_guid, comment["text"], comment["comment_type"]): + self._load_comments() + + self.app.push_screen(AddCommunityCommentScreen(self.community_name), callback=on_comment) diff --git a/my_egeria/my_egeria/DemoCode/My_Profile/MainScreen.py b/my_egeria/my_egeria/DemoCode/My_Profile/MainScreen.py index 0b59d401..9e483775 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("[] Work with Communities"), Option("[] Change User"), Option("[] Leave Feedback"), Option("[] Feedback Log"), 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 ee843555..6f77a06c 100644 --- a/my_egeria/my_egeria/DemoCode/My_Profile/RETURN_CODES.md +++ b/my_egeria/my_egeria/DemoCode/My_Profile/RETURN_CODES.md @@ -12,9 +12,9 @@ In accordance with application conventions: | Return Code | Name / Type | Description | Source Components | |---|---|---|---| -| **`200`** | `SUCCESS_OK` / `DEFAULT_EXIT` | Standard success code. Used for successful operations, standard modal dismissal, backing out of dialogs, cancelling inputs without error, or quitting cleanly. | `my_egeria.py`, `MainScreen.py`, `StatusScreen.py`, `CreateProfileScreen.py`, `EditElementsScreens.py`, `AddToElementsScreens.py`, `TechnologyTypeScreens.py`, `tech_types_handler.py`, `shop_for_data_handler.py`, `team_roles_handler.py`, `SelectionOverviewScreen.py`, `GenericDataViewScreen.py`, `MyBookMarksScreen.py`, `MyTeamScreen.py`, `SearchForTermScreen.py`, `ShowCommentScreen.py` | +| **`200`** | `SUCCESS_OK` / `DEFAULT_EXIT` | Standard success code. Used for successful operations, standard modal dismissal, backing out of dialogs, cancelling inputs without error, or quitting cleanly. | `my_egeria.py`, `MainScreen.py`, `StatusScreen.py`, `CreateProfileScreen.py`, `EditElementsScreens.py`, `AddToElementsScreens.py`, `TechnologyTypeScreens.py`, `tech_types_handler.py`, `shop_for_data_handler.py`, `team_roles_handler.py`, `SelectionOverviewScreen.py`, `GenericDataViewScreen.py`, `MyBookMarksScreen.py`, `MyTeamScreen.py`, `SearchForTermScreen.py`, `ShowCommentScreen.py`, `MyNetworkScreen.py`, `CommunitiesScreen.py`, `CommunityScreen.py` | | **`201`** | `ALT_NAVIGATION` / `NO_MATCH` | Alternative navigation or fallback action code. In search/team screens, indicates no matching terms/roles found and triggers transition to data shopping or alternate view. In tech type selection, represents secondary navigation action. | `SearchForTermScreen.py`, `team_roles_handler.py`, `TechnologyTypeScreens.py`, `MyTeamScreen.py`, `ShopForDataScreen.py` | -| **`210`** | `QUIT_TO_MAIN` | Exit/Quit action code from data shopping / data viewing screens to dismiss the view and return directly to the main menu screen. | `ShopForDataScreen.py`, `GenericDataViewScreen.py`, `SelectionOverviewScreen.py`, `shop_for_data_handler.py` | +| **`210`** | `QUIT_TO_MAIN` | Exit/Quit action code from data shopping / data viewing screens, and from a single community's screen, to dismiss the view and return directly to the main menu screen. | `ShopForDataScreen.py`, `GenericDataViewScreen.py`, `SelectionOverviewScreen.py`, `shop_for_data_handler.py`, `CommunityScreen.py`, `CommunitiesScreen.py` | | **`211`** | `SUBSCRIBE_ACTION` | Action code indicating the user has requested to create a digital subscription for the selected data source, catalog product, or data element. | `ShopForDataScreen.py`, `GenericDataViewScreen.py`, `SelectionOverviewScreen.py`, `shop_for_data_handler.py` | | **`212`** | `SAMPLE_DATA_ACTION` | Action code indicating the user has requested to view sample data for the selected data source in Shop for Data. | `ShopForDataScreen.py`, `shop_for_data_handler.py` | diff --git a/my_egeria/my_egeria/DemoCode/My_Profile/community_handler.py b/my_egeria/my_egeria/DemoCode/My_Profile/community_handler.py new file mode 100644 index 00000000..3713e661 --- /dev/null +++ b/my_egeria/my_egeria/DemoCode/My_Profile/community_handler.py @@ -0,0 +1,344 @@ +""" + PDX-License-Identifier: Apache-2.0 + Copyright Contributors to the ODPi Egeria project. + + Work with Communities handler mixin for My Profile Textual App. + + A person belongs to a community when an actor is linked to it by an + AssignmentScope relationship, whose assignmentType says how they take part + (https://egeria-project.org/types/1/0120-Assignment-Scopes/). The actor is + either the person's own profile, or a role they are appointed to. + + Joining links the user's profile to the community with link_assignment_scope, + and leaving removes that link with detach_assignment_scope. Memberships held + through a role are listed but not removed here: the role may be shared, and + changing it is a job for whoever manages the role. + + A community's blog is a NoteLog attached to the community, holding BlogEntry + notes. The app creates the log (qualified name BLOG_LOG_QN) on the first post; + posts are read from every note log attached to the community, so entries + written by other tools are shown too. Comments are attached to the community + directly. +""" + +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 CommunitiesScreen import CommunitiesScreen, DIRECT_LINK, JOIN_ASSIGNMENT_TYPES +from CommunityScreen import comment_type_label + +COMMUNITY_TYPE = "Community" +BLOG_LOG_QN = "NoteLog::{community_guid}::CommunityBlog" +# How many of a community's most recent blog posts are listed +LATEST_POSTS = 20 + + +def _type_name(element: dict) -> str: + return ((element.get("elementHeader") or {}).get("type") or {}).get("typeName") or "" + + +def _guid(element: dict) -> str: + return (element.get("elementHeader") or {}).get("guid") or "" + + +def _name(properties: dict) -> str: + return properties.get("displayName") or properties.get("fullName") or properties.get("qualifiedName") or "" + + +def _created(element: dict) -> tuple[str, str]: + """(created by, create time) from an element's header; the time is ISO 8601 text.""" + versions = (element.get("elementHeader") or {}).get("versions") or {} + return versions.get("createdBy") or "", str(versions.get("createTime") or "") + + +def _posted(create_time: str) -> str: + """'2026-10-23T13:27:27.750+00:00' -> '2026-10-23 13:27'""" + return create_time[:16].replace("T", " ") + + +class CommunitiesMixin: + """Mixin class providing the Work with Communities feature for MyProfileApp.""" + + def _community_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_communities(self) -> list[tuple[str, str, str, str, str]] | None: + """The user's communities as (name, assignment type, linked through, description, GUID) rows, + or None if Egeria could not be reached. A community reached through several links is one row.""" + client = None + try: + client = self._community_client() + profile = client.get_my_profile(output_format="JSON") + except PyegeriaException as e: + self.log(f"Retrieving my communities failed: {e}") + self.notify(f"Could not retrieve your communities: {e}", timeout=10, severity="error") + return None + finally: + if client: + client.close_session() + if not isinstance(profile, dict): + return [] + + # (link, linked through) pairs: links from the profile itself, then links from its roles + links = [(link, DIRECT_LINK) for link in profile.get("assignmentScope") or []] + for appointment in profile.get("performsRoles") or []: + role = appointment.get("relatedElement") or {} + role_name = _name(role.get("properties") or {}) + links += [(link, f"Role: {role_name}") for link in appointment.get("nestedElements") or []] + + communities: dict[str, dict] = {} + for link, linked_through in links: + community = link.get("relatedElement") or {} + guid = _guid(community) + if _type_name(community) != COMMUNITY_TYPE or not guid: + continue + properties = community.get("properties") or {} + entry = communities.setdefault(guid, {"name": _name(properties), + "description": properties.get("description") or "", + "types": [], "through": []}) + assignment_type = (link.get("relationshipProperties") or {}).get("assignmentType") or "" + if assignment_type and assignment_type not in entry["types"]: + entry["types"].append(assignment_type) + if linked_through not in entry["through"]: + entry["through"].append(linked_through) + return [(c["name"], ", ".join(c["types"]), "; ".join(c["through"]), c["description"], guid) + for guid, c in communities.items()] + + def show_communities(self) -> None: + """Show the Work with Communities screen for the current user.""" + if not getattr(self, "user_GUID", ""): + self.notify("Your profile has not loaded yet, so your communities can't be shown", timeout=10, + severity="warning") + return + communities = self.list_my_communities() + if communities is None: + return + self.push_screen(CommunitiesScreen(communities), callback=self.communities_callback) + + def communities_callback(self, result: Any) -> None: + self.log(f"Communities screen returned: {result}") + + def search_communities(self, search_string: str) -> list[tuple[str, str, str, str]] | None: + """Communities matching search_string ('*' for all) as (name, description, qualified name, GUID) rows, + or None if Egeria could not be reached.""" + client = None + try: + client = self._community_client() + # Depth 0: only the communities' own properties are listed; members are fetched on request + found = client.find_communities(search_string=search_string or "*", ignore_case=True, + output_format="JSON", graph_query_depth=0) + except PyegeriaException as e: + self.log(f"Searching for communities matching {search_string} failed: {e}") + self.notify(f"Searching for communities failed: {e}", timeout=10, severity="error") + return None + finally: + if client: + client.close_session() + self.log(f"Communities found for {search_string}: {found}") + rows = [] + # "No elements found" is returned, rather than an exception, when nothing matches + for community in found if isinstance(found, list) else []: + guid = _guid(community) + if not guid: + continue + properties = community.get("properties") or {} + rows.append((_name(properties), properties.get("description") or "", + properties.get("qualifiedName") or "", guid)) + return rows + + def get_community_members(self, community_guid: str) -> list[tuple[str, str, str, str, str]] | None: + """The actors linked to a community as (name, actor type, assignment type, qualified name, GUID) rows, + or None if they could not be retrieved. An actor may be a person or a role.""" + client = None + try: + client = self._community_client() + community = client.get_community_by_guid(community_guid, output_format="JSON", graph_query_depth=1) + except PyegeriaException as e: + self.log(f"Retrieving members of community {community_guid} failed: {e}") + self.notify(f"Could not retrieve the community's members: {e}", timeout=10, severity="error") + return None + finally: + if client: + client.close_session() + self.log(f"Community {community_guid}: {community}") + rows = [] + for link in (community.get("assignedActors") if isinstance(community, dict) else None) or []: + actor = link.get("relatedElement") or {} + properties = actor.get("properties") or {} + rows.append((_name(properties), _type_name(actor), + (link.get("relationshipProperties") or {}).get("assignmentType") or "", + properties.get("qualifiedName") or "", _guid(actor))) + return rows + + def get_community_blog_posts(self, community_guid: str) -> list[tuple[str, str, str, str, str]] | None: + """The community's latest blog posts, newest first, as (title, author, posted, text, GUID) rows, + or None if they could not be retrieved.""" + client = None + posts = [] + try: + client = self._community_client() + logs = client.get_attached_note_logs(community_guid, graph_query_depth=0) + # "No elements found" is returned, rather than an exception, when there are none + for log in logs if isinstance(logs, list) else []: + entries = client.get_notes_for_note_log(_guid(log), metadata_element_type_name="BlogEntry", + graph_query_depth=0) + posts += entries if isinstance(entries, list) else [] + except PyegeriaException as e: + self.log(f"Retrieving the blog of community {community_guid} failed: {e}") + self.notify(f"Could not retrieve the community's blog: {e}", timeout=10, severity="error") + return None + finally: + if client: + client.close_session() + rows = [] + for post in posts: + properties = post.get("properties") or {} + author, create_time = _created(post) + rows.append((create_time, (_name(properties), author, _posted(create_time), + properties.get("description") or "", _guid(post)))) + rows.sort(key=lambda r: r[0], reverse=True) + return [row for _, row in rows[:LATEST_POSTS]] + + def _community_blog_log(self, client: Egeria, community_guid: str, community_name: str) -> str: + """The GUID of the community's blog note log, which is created on first use.""" + qualified_name = BLOG_LOG_QN.format(community_guid=community_guid) + logs = client.get_attached_note_logs(community_guid, graph_query_depth=0) + for log in logs if isinstance(logs, list) else []: + if (log.get("properties") or {}).get("qualifiedName") == qualified_name: + return _guid(log) + body = { + "class": "NewAttachmentRequestBody", + "properties": { + "class": "NoteLogProperties", + "typeName": "NoteLog", + "qualifiedName": qualified_name, + "displayName": f"{community_name} Blog", + "description": f"Blog for the members of {community_name}.", + } + } + log_guid = client.create_note_log(element_guid=community_guid, body=body) + self.log(f"Created blog for community {community_guid}: {log_guid}") + return log_guid + + def add_community_blog_post(self, community_guid: str, community_name: str, title: str, text: str) -> bool: + """Post to the community's blog, creating the blog if it doesn't exist yet. Returns True on success.""" + client = None + try: + client = self._community_client() + log_guid = self._community_blog_log(client, community_guid, community_name) + properties = { + "class": "BlogEntryProperties", + "typeName": "BlogEntry", + "qualifiedName": client.make_feedback_qn("BlogEntry", community_guid, title), + "displayName": title, + "description": text, + } + post_guid = client.create_note(log_guid, body=properties) + self.log(f"Posted to the blog of community {community_guid}: {post_guid}") + self.notify(f"Your post '{title}' has been added to the {community_name} blog", timeout=10, + severity="information") + return True + except PyegeriaException as e: + self.log(f"Posting to the blog of community {community_guid} failed: {e}") + self.notify(f"Posting to the community's blog failed: {e}", timeout=10, severity="error") + return False + finally: + if client: + client.close_session() + + def get_community_comments(self, community_guid: str) -> list[tuple[str, str, str, str, str]] | None: + """The comments attached to the community, newest first, as (comment, type, author, posted, GUID) rows, + or None if they could not be retrieved.""" + client = None + try: + client = self._community_client() + comments = client.get_attached_comments(community_guid, graph_query_depth=0) + except PyegeriaException as e: + self.log(f"Retrieving the comments on community {community_guid} failed: {e}") + self.notify(f"Could not retrieve the community's comments: {e}", timeout=10, severity="error") + return None + finally: + if client: + client.close_session() + rows = [] + for comment in comments if isinstance(comments, list) else []: + properties = comment.get("properties") or {} + author, create_time = _created(comment) + comment_type = comment_type_label(properties.get("commentType") or "") + rows.append((create_time, (properties.get("description") or "", comment_type, author, + _posted(create_time), _guid(comment)))) + rows.sort(key=lambda r: r[0], reverse=True) + return [row for _, row in rows] + + def add_community_comment(self, community_guid: str, text: str, comment_type: str) -> bool: + """Attach a comment to the community. comment_type is one of pyegeria's COMMENT_TYPES. + Returns True on success.""" + client = None + try: + client = self._community_client() + response = client.add_comment_to_element(community_guid, text, comment_type) + self.log(f"Commented on community {community_guid}: {response}") + self.notify("Your comment has been added", timeout=10, severity="information") + return True + except PyegeriaException as e: + self.log(f"Commenting on community {community_guid} failed: {e}") + self.notify(f"Adding the comment failed: {e}", timeout=10, severity="error") + return False + finally: + if client: + client.close_session() + + def join_community(self, community_guid: str, assignment_type: str = JOIN_ASSIGNMENT_TYPES[0]) -> bool: + """Link the user's profile to the community. Returns True on success.""" + body = { + "class": "NewRelationshipRequestBody", + "properties": { + "class": "AssignmentScopeProperties", + "assignmentType": assignment_type, + } + } + client = None + try: + client = self._community_client() + client.link_assignment_scope(community_guid, self.user_GUID, body) + self.notify(f"You have joined the community as a {assignment_type}", timeout=10, + severity="information") + return True + except PyegeriaException as e: + self.log(f"Joining community {community_guid} failed: {e}") + self.notify(f"Joining the community failed: {e}", timeout=10, severity="error") + return False + finally: + if client: + client.close_session() + + def leave_community(self, community_guid: str) -> bool: + """Remove the link between the user's profile and the community. Returns True on success.""" + client = None + try: + client = self._community_client() + # same GUID order as link_assignment_scope: scope (the community) first, actor second + client.detach_assignment_scope(community_guid, self.user_GUID, {"class": "DeleteRelationshipRequestBody"}) + self.notify("You have left the community", timeout=10, severity="information") + return True + except PyegeriaException as e: + self.log(f"Leaving community {community_guid} failed: {e}") + self.notify(f"Leaving the community failed: {e}", timeout=10, severity="error") + return False + finally: + if client: + client.close_session() 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 e26ed355..6d0f4838 100644 --- a/my_egeria/my_egeria/DemoCode/My_Profile/my_egeria.py +++ b/my_egeria/my_egeria/DemoCode/My_Profile/my_egeria.py @@ -94,10 +94,11 @@ from bookmarks_handler import BookmarksMixin, bookmarks_qualified_name from MyBookMarksScreen import MyBookMarksScreen from network_handler import NetworkMixin +from community_handler import CommunitiesMixin class MyProfileApp(App, TechTypesMixin, ShopForDataMixin, TeamRolesMixin, ElementsCrudMixin, FeedbackMixin, - BookmarksMixin, NetworkMixin): + BookmarksMixin, NetworkMixin, CommunitiesMixin): """My Profile App. Retrieves a user's profile from Egeria and displays current work items. @@ -720,6 +721,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 == "Work with Communities": + self.show_communities() elif selected_option == "Change User": await self.push_screen(ChangeUserScreen(), callback=self.change_user_callback) elif selected_option == "Subscriptions": 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 21cf56f2..ef898b66 100644 --- a/my_egeria/my_egeria/DemoCode/My_Profile/my_profile.tcss +++ b/my_egeria/my_egeria/DemoCode/My_Profile/my_profile.tcss @@ -103,11 +103,98 @@ MyNetworkScreen { max-height: 12; } +/* Title row, then my communities + find side by side */ +CommunitiesScreen { + grid-rows: auto 1fr; + background: $background; +} + +#my_communities_container { + column-span: 2; + width: 100%; + height: 100%; +} + +#find_communities_container { + width: 100%; + height: 100%; +} + +#my_communities_table { + width: 100%; + height: 1fr; +} + +#my_communities_buttons, +#found_communities_buttons { + height: auto; +} + +#join_assignment_type { + width: 20; +} + +#communities_found_table { + height: auto; + max-height: 12; +} + +/* Title row, then the blog + members side by side, then the comments */ +CommunityScreen { + grid-rows: auto 1fr 1fr; + background: $background; +} + +#community_blog_container { + column-span: 2; + width: 100%; + height: 100%; +} + +#community_members_container { + width: 100%; + height: 100%; +} + +#community_comments_container { + column-span: 3; + width: 100%; + height: 100%; +} + +#community_posts_table, +#community_members_table, +#community_comments_table { + height: auto; + max-height: 12; +} + +#community_post_text { + padding: 1 0; +} + +NewBlogPostScreen, AddCommunityCommentScreen { + grid-rows: auto 1fr; + background: $background; +} + +#new_blog_post_container, #add_community_comment_container { + column-span: 3; +} + +#new_blog_post_container Horizontal, #add_community_comment_container Horizontal { + height: auto; +} + +#blog_post_text, #community_comment_text { + height: 10; +} + ModalScreen > Header, ModalScreen > Footer { column-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 { +#communities_title, #community_title, #new_blog_post_title, #add_community_comment_title, #confirm_leave_community_title, #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; 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 fa6c0ef7..d36bc8e8 100644 --- a/my_egeria/my_egeria/DemoCode/My_Profile/profile_utils.py +++ b/my_egeria/my_egeria/DemoCode/My_Profile/profile_utils.py @@ -5,7 +5,6 @@ Utility and data transformation functions for My Profile Textual App. """ -import json import re import sys from typing import Any @@ -37,33 +36,27 @@ def check_request_serialization() -> str | None: 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. + The client constructor checks that the platform answers, and the token request + checks the credentials: create_egeria_bearer_token() raises on a 401/403. + The messages are kept to one line, as they are shown on a screen; str() of a + pyegeria exception is a multi-line diagnostic dump. """ - from pyegeria import Egeria, PyegeriaConnectionException, PyegeriaException + from pyegeria import (Egeria, PyegeriaConnectionException, PyegeriaException, + PyegeriaTimeoutException, PyegeriaUnauthorizedException) 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: + eclient.create_egeria_bearer_token(user_name, user_password) + except (PyegeriaConnectionException, PyegeriaTimeoutException): return f"Unable to connect to the Egeria platform at {platform_url}." + except PyegeriaUnauthorizedException: + return f"Egeria rejected the credentials for user {user_name}." except PyegeriaException as e: - return f"Egeria at {platform_url} returned an error: {e!s}" + return f"Egeria at {platform_url} returned an error: {getattr(e, 'message', '') or type(e).__name__}" 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 diff --git a/tests/micro-tests/my_profile/test_community_handler.py b/tests/micro-tests/my_profile/test_community_handler.py new file mode 100644 index 00000000..fc3c4e4d --- /dev/null +++ b/tests/micro-tests/my_profile/test_community_handler.py @@ -0,0 +1,593 @@ +""" + PDX-License-Identifier: Apache-2.0 + Copyright Contributors to the ODPi Egeria project. + + Unit tests for the Work with Communities mixin and screens (a user belongs to a + community when their profile, or a role they hold, is linked to it by an + AssignmentScope relationship; a community has a blog note log and comments). +""" + +from unittest.mock import MagicMock, patch + +import pytest +from textual.app import App +from textual.widgets import DataTable, Input, Button, Select, Static, TextArea + +import community_handler +from community_handler import CommunitiesMixin +from CommunitiesScreen import CommunitiesScreen, ConfirmLeaveCommunityScreen, DIRECT_LINK +from CommunityScreen import CommunityScreen, NewBlogPostScreen, AddCommunityCommentScreen +from pyegeria import PyegeriaException + + +def _element(guid, type_name, display_name, description=""): + return {"elementHeader": {"guid": guid, "type": {"typeName": type_name}}, + "properties": {"displayName": display_name, "qualifiedName": f"{type_name}::{guid}", + "description": description}} + + +def _created(element, by, at): + element["elementHeader"]["versions"] = {"createdBy": by, "createTime": at} + return element + + +def _scope(element, assignment_type): + return {"relationshipHeader": {"guid": f"rel-{element['elementHeader']['guid']}"}, + "relationshipProperties": {"class": "AssignmentScopeProperties", "assignmentType": assignment_type}, + "relatedElement": element} + + +DATA_SCI = _element("ds-guid", "Community", "Data Science SIG", "Data science") +SUSTAIN = _element("sus-guid", "Community", "Sustainability Community", "Green IT") + +PROFILE = { + "elementHeader": {"guid": "profile-guid"}, + "assignmentScope": [ + _scope(DATA_SCI, "Contributor"), + _scope(_element("proj-guid", "Project", "A project"), "Owner"), # not a community + ], + "performsRoles": [ + {"relatedElement": _element("dc-role", "GovernanceRole", "DC Leader"), + "nestedElements": [_scope(SUSTAIN, "Contributor")]}, + {"relatedElement": _element("champ-role", "GovernanceRole", "IT Champion"), + "nestedElements": [_scope(SUSTAIN, "Contributor"), + _scope(_element("loc-guid", "Location", "London"), None)]}, + {"relatedElement": _element("tester-role", "PersonRole", "Tester")}, + ], +} + + +class DummyApp(CommunitiesMixin): + 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("community_handler.Egeria") as mock_egeria: + yield mock_egeria.return_value + + +class TestCommunitiesMixin: + def test_list_merges_direct_and_role_memberships(self, client): + client.get_my_profile.return_value = PROFILE + rows = DummyApp().list_my_communities() + client.get_my_profile.assert_called_once_with(output_format="JSON") + # One row per community; non-communities are skipped + assert rows == [ + ("Data Science SIG", "Contributor", DIRECT_LINK, "Data science", "ds-guid"), + ("Sustainability Community", "Contributor", "Role: DC Leader; Role: IT Champion", "Green IT", "sus-guid"), + ] + client.close_session.assert_called_once() + + def test_list_without_profile_is_empty(self, client): + client.get_my_profile.return_value = "No elements found" + assert DummyApp().list_my_communities() == [] + + def test_list_error_returns_none(self, client): + client.get_my_profile.side_effect = PyegeriaException("boom") + app = DummyApp() + assert app.list_my_communities() is None + assert app.notices[-1][1] == "error" + + def test_show_pushes_screen(self, client): + client.get_my_profile.return_value = PROFILE + app = DummyApp() + app.show_communities() + screen, callback = app.pushed[-1] + assert isinstance(screen, CommunitiesScreen) + assert screen.my_communities_data[0][4] == "ds-guid" + + def test_show_without_profile_is_refused(self, client): + app = DummyApp() + app.user_GUID = "" + app.show_communities() + assert app.pushed == [] + client.get_my_profile.assert_not_called() + assert app.notices[-1][1] == "warning" + + def test_search(self, client): + client.find_communities.return_value = [DATA_SCI, SUSTAIN] + rows = DummyApp().search_communities("*") + assert client.find_communities.call_args.kwargs["search_string"] == "*" + assert rows == [("Data Science SIG", "Data science", "Community::ds-guid", "ds-guid"), + ("Sustainability Community", "Green IT", "Community::sus-guid", "sus-guid")] + + def test_empty_search_means_all(self, client): + client.find_communities.return_value = "No elements found" + assert DummyApp().search_communities("") == [] + assert client.find_communities.call_args.kwargs["search_string"] == "*" + + def test_search_error_returns_none(self, client): + client.find_communities.side_effect = PyegeriaException("boom") + assert DummyApp().search_communities("sig") is None + + def test_members_are_people_and_roles(self, client): + client.get_community_by_guid.return_value = { + "assignedActors": [_scope(_element("gary-guid", "Person", "Gary"), "Contributor"), + _scope(_element("lead-role", "PersonRole", "SIG Leader"), "Leader")]} + rows = DummyApp().get_community_members("ds-guid") + assert client.get_community_by_guid.call_args.args == ("ds-guid",) + assert rows == [("Gary", "Person", "Contributor", "Person::gary-guid", "gary-guid"), + ("SIG Leader", "PersonRole", "Leader", "PersonRole::lead-role", "lead-role")] + + def test_members_error_returns_none(self, client): + client.get_community_by_guid.side_effect = PyegeriaException("boom") + app = DummyApp() + assert app.get_community_members("ds-guid") is None + assert app.notices[-1][1] == "error" + + def test_join(self, client): + assert DummyApp().join_community("ds-guid", "Observer") is True + scope, actor, body = client.link_assignment_scope.call_args.args + # The community is the scope, the user's profile the actor + assert (scope, actor) == ("ds-guid", "profile-guid") + assert body["properties"] == {"class": "AssignmentScopeProperties", "assignmentType": "Observer"} + + def test_join_error(self, client): + client.link_assignment_scope.side_effect = PyegeriaException("boom") + app = DummyApp() + assert app.join_community("ds-guid") is False + assert app.notices[-1][1] == "error" + + def test_leave(self, client): + assert DummyApp().leave_community("ds-guid") is True + client.detach_assignment_scope.assert_called_once_with( + "ds-guid", "profile-guid", {"class": "DeleteRelationshipRequestBody"}) + + def test_leave_error(self, client): + client.detach_assignment_scope.side_effect = PyegeriaException("boom") + app = DummyApp() + assert app.leave_community("ds-guid") is False + assert app.notices[-1][1] == "error" + + def test_blog_posts_from_every_attached_log_newest_first(self, client): + client.get_attached_note_logs.return_value = [_element("blog-log", "NoteLog", "Blog"), + _element("other-log", "NoteLog", "Notifications")] + old = _created(_element("p1", "BlogEntry", "First", "Hello"), "gary", "2026-10-01T09:00:00.000+00:00") + new = _created(_element("p2", "BlogEntry", "Second", "Again"), "erin", "2026-10-20T13:27:27.750+00:00") + client.get_notes_for_note_log.side_effect = [[old], [new]] + rows = DummyApp().get_community_blog_posts("ds-guid") + assert client.get_attached_note_logs.call_args.args == ("ds-guid",) + # only blog entries are asked for, so notifications in other logs are skipped + for call in client.get_notes_for_note_log.call_args_list: + assert call.kwargs["metadata_element_type_name"] == "BlogEntry" + assert rows == [("Second", "erin", "2026-10-20 13:27", "Again", "p2"), + ("First", "gary", "2026-10-01 09:00", "Hello", "p1")] + + def test_blog_posts_are_limited_to_the_latest(self, client): + client.get_attached_note_logs.return_value = [_element("blog-log", "NoteLog", "Blog")] + client.get_notes_for_note_log.return_value = [ + _created(_element(f"p{i:02}", "BlogEntry", f"Post {i}"), "gary", f"2026-10-{i:02}T09:00:00") + for i in range(1, 26)] + rows = DummyApp().get_community_blog_posts("ds-guid") + assert len(rows) == community_handler.LATEST_POSTS + assert rows[0][4] == "p25" + + def test_no_blog_yet(self, client): + client.get_attached_note_logs.return_value = "No elements found" + assert DummyApp().get_community_blog_posts("ds-guid") == [] + client.get_notes_for_note_log.assert_not_called() + + def test_blog_error_returns_none(self, client): + client.get_attached_note_logs.side_effect = PyegeriaException("boom") + app = DummyApp() + assert app.get_community_blog_posts("ds-guid") is None + assert app.notices[-1][1] == "error" + + def test_first_post_creates_the_blog(self, client): + client.get_attached_note_logs.return_value = "No elements found" + client.create_note_log.return_value = "new-log" + client.make_feedback_qn.return_value = "BlogEntry::ds-guid::garygeeke::Hi::1" + assert DummyApp().add_community_blog_post("ds-guid", "Data Science SIG", "Hi", "Hello all") is True + body = client.create_note_log.call_args.kwargs["body"] + assert client.create_note_log.call_args.kwargs["element_guid"] == "ds-guid" + assert body["class"] == "NewAttachmentRequestBody" + assert body["properties"]["qualifiedName"] == "NoteLog::ds-guid::CommunityBlog" + log_guid = client.create_note.call_args.args[0] + properties = client.create_note.call_args.kwargs["body"] + assert log_guid == "new-log" + assert properties["typeName"] == "BlogEntry" + assert (properties["displayName"], properties["description"]) == ("Hi", "Hello all") + + def test_later_posts_reuse_the_blog(self, client): + blog = _element("blog-log", "NoteLog", "Blog") + blog["properties"]["qualifiedName"] = "NoteLog::ds-guid::CommunityBlog" + client.get_attached_note_logs.return_value = [_element("other-log", "NoteLog", "Notifications"), blog] + assert DummyApp().add_community_blog_post("ds-guid", "Data Science SIG", "Hi", "Hello") is True + client.create_note_log.assert_not_called() + assert client.create_note.call_args.args[0] == "blog-log" + + def test_post_error(self, client): + client.get_attached_note_logs.return_value = "No elements found" + client.create_note_log.side_effect = PyegeriaException("boom") + app = DummyApp() + assert app.add_community_blog_post("ds-guid", "Data Science SIG", "Hi", "Hello") is False + client.create_note.assert_not_called() + assert app.notices[-1][1] == "error" + + def test_comments_newest_first(self, client): + def comment(guid, text, comment_type, at): + element = _created(_element(guid, "Comment", ""), "gary", at) + element["properties"].update({"description": text, "commentType": comment_type}) + return element + client.get_attached_comments.return_value = [ + comment("c1", "Why?", "QUESTION", "2026-10-01T09:00:00"), + comment("c2", "Because", "USAGE_EXPERIENCE", "2026-10-02T09:00:00")] + rows = DummyApp().get_community_comments("ds-guid") + assert client.get_attached_comments.call_args.args == ("ds-guid",) + assert rows == [("Because", "Usage Experience", "gary", "2026-10-02 09:00", "c2"), + ("Why?", "Question", "gary", "2026-10-01 09:00", "c1")] + + def test_no_comments(self, client): + client.get_attached_comments.return_value = "No elements found" + assert DummyApp().get_community_comments("ds-guid") == [] + + def test_comments_error_returns_none(self, client): + client.get_attached_comments.side_effect = PyegeriaException("boom") + assert DummyApp().get_community_comments("ds-guid") is None + + def test_add_comment(self, client): + assert DummyApp().add_community_comment("ds-guid", "Nice", "SUGGESTION") is True + client.add_comment_to_element.assert_called_once_with("ds-guid", "Nice", "SUGGESTION") + + def test_add_comment_error(self, client): + client.add_comment_to_element.side_effect = PyegeriaException("boom") + app = DummyApp() + assert app.add_community_comment("ds-guid", "Nice", "SUGGESTION") is False + assert app.notices[-1][1] == "error" + + +MY_ROWS = [("Data Science SIG", "Contributor", DIRECT_LINK, "Data science", "ds-guid"), + ("Sustainability Community", "Contributor", "Role: DC Leader", "Green IT", "sus-guid")] +ALL_COMMUNITIES = [("Data Science SIG", "Data science", "Community::ds-guid", "ds-guid"), + ("Sustainability Community", "Green IT", "Community::sus-guid", "sus-guid"), + ("Privacy Community", "Privacy", "Community::priv-guid", "priv-guid")] + + +class CommunitiesHostApp(App): + """Host app standing in for MyProfileApp's community methods.""" + + def __init__(self, rows): + super().__init__() + self.rows = list(rows) + self.join_community = MagicMock(side_effect=self._join) + self.leave_community = MagicMock(side_effect=self._leave) + self.search_communities = MagicMock(return_value=list(ALL_COMMUNITIES)) + self.get_community_members = MagicMock(return_value=[("Gary", "Person", "Contributor", "Person::g", "g")]) + self.posts = [("Second", "erin", "2026-10-20 13:27", "Again", "p2"), + ("First", "gary", "2026-10-01 09:00", "Hello", "p1")] + self.comments = [("Why?", "Question", "gary", "2026-10-01 09:00", "c1")] + self.get_community_blog_posts = MagicMock(side_effect=lambda guid: list(self.posts)) + self.get_community_comments = MagicMock(side_effect=lambda guid: list(self.comments)) + self.add_community_blog_post = MagicMock(side_effect=self._post) + self.add_community_comment = MagicMock(side_effect=self._comment) + self.dismissed = None + + def _post(self, guid, name, title, text): + self.posts.insert(0, (title, "garygeeke", "2026-10-28 12:00", text, "p-new")) + return True + + def _comment(self, guid, text, comment_type): + self.comments.insert(0, (text, comment_type, "garygeeke", "2026-10-28 12:00", "c-new")) + return True + + def _join(self, guid, assignment_type="Contributor"): + name = next(r[0] for r in ALL_COMMUNITIES if r[3] == guid) + self.rows.append((name, assignment_type, DIRECT_LINK, "", guid)) + return True + + def _leave(self, guid): + self.rows = [r for r in self.rows if r[4] != guid] + return True + + def list_my_communities(self): + return list(self.rows) + + async def on_mount(self): + self.screen_under_test = CommunitiesScreen(list(self.rows)) + await self.push_screen(self.screen_under_test, callback=lambda r: setattr(self, "dismissed", r)) + + +class TestCommunitiesScreen: + @pytest.mark.asyncio + async def test_lists_mine_and_finds_all_others(self): + app = CommunitiesHostApp(MY_ROWS) + async with app.run_test(): + screen = app.screen_under_test + assert screen.query_one("#my_communities_table", DataTable).row_count == 2 + # '*' is searched on opening, and communities the user is already in are left out + app.search_communities.assert_called_once_with("*") + found = screen.query_one("#communities_found_table", DataTable) + assert found.row_count == 1 + assert found.get_row_at(0)[3] == "priv-guid" + + @pytest.mark.asyncio + async def test_search_term(self): + app = CommunitiesHostApp(MY_ROWS) + async with app.run_test() as pilot: + search = app.screen_under_test.query_one("#community_search_input", Input) + search.value = "privacy" + await search.action_submit() + await pilot.pause() + app.search_communities.assert_called_with("privacy") + + @pytest.mark.asyncio + async def test_open_my_community(self): + app = CommunitiesHostApp(MY_ROWS) + async with app.run_test() as pilot: + screen = app.screen_under_test + screen.query_one("#my_communities_table", DataTable).move_cursor(row=1) + screen.query_one("#open_my_community_btn", Button).press() + await pilot.pause() + assert isinstance(app.screen, CommunityScreen) + assert app.screen.community_guid == "sus-guid" + app.get_community_members.assert_called_once_with("sus-guid") + + @pytest.mark.asyncio + async def test_open_found_community(self): + app = CommunitiesHostApp(MY_ROWS) + async with app.run_test() as pilot: + screen = app.screen_under_test + screen.query_one("#communities_found_table", DataTable).focus() + await pilot.press("ctrl+o") + await pilot.pause() + assert isinstance(app.screen, CommunityScreen) + assert app.screen.community_guid == "priv-guid" + + @pytest.mark.asyncio + async def test_back_from_a_community_returns_to_the_list(self): + app = CommunitiesHostApp(MY_ROWS) + async with app.run_test() as pilot: + app.screen_under_test.action_open_community() + await pilot.pause() + await pilot.press("escape") + await pilot.pause() + assert app.screen is app.screen_under_test + assert app.dismissed is None + + @pytest.mark.asyncio + async def test_q_from_a_community_goes_to_the_main_screen(self): + app = CommunitiesHostApp(MY_ROWS) + async with app.run_test() as pilot: + app.screen_under_test.action_open_community() + await pilot.pause() + await pilot.press("q") + await pilot.pause() + assert app.dismissed == 200 + + @pytest.mark.asyncio + async def test_join_with_chosen_assignment_type(self): + app = CommunitiesHostApp(MY_ROWS) + async with app.run_test() as pilot: + screen = app.screen_under_test + screen.query_one("#join_assignment_type", Select).value = "Observer" + screen.query_one("#join_community_btn", Button).press() + await pilot.pause() + app.join_community.assert_called_once_with("priv-guid", "Observer") + assert screen.query_one("#my_communities_table", DataTable).row_count == 3 + assert screen.query_one("#communities_found_table", DataTable).row_count == 0 + + @pytest.mark.asyncio + async def test_join_with_nothing_found(self): + app = CommunitiesHostApp(MY_ROWS) + app.search_communities.return_value = [] + async with app.run_test() as pilot: + app.screen_under_test.action_join_community() + await pilot.pause() + app.join_community.assert_not_called() + + @pytest.mark.asyncio + async def test_leave_asks_for_confirmation(self): + app = CommunitiesHostApp(MY_ROWS) + async with app.run_test() as pilot: + screen = app.screen_under_test + screen.action_leave_community() + await pilot.pause() + assert isinstance(app.screen, ConfirmLeaveCommunityScreen) + app.screen.query_one("#cancel_leave_community_btn", Button).press() + await pilot.pause() + app.leave_community.assert_not_called() + assert screen.query_one("#my_communities_table", DataTable).row_count == 2 + + @pytest.mark.asyncio + async def test_leave_confirmed_offers_the_community_again(self): + app = CommunitiesHostApp(MY_ROWS) + async with app.run_test() as pilot: + screen = app.screen_under_test + screen.action_leave_community() + await pilot.pause() + app.screen.query_one("#confirm_leave_community_btn", Button).press() + await pilot.pause() + app.leave_community.assert_called_once_with("ds-guid") + assert screen.query_one("#my_communities_table", DataTable).row_count == 1 + found = screen.query_one("#communities_found_table", DataTable) + assert "ds-guid" in {found.get_row_at(i)[3] for i in range(found.row_count)} + + @pytest.mark.asyncio + async def test_membership_through_a_role_cannot_be_left_here(self): + app = CommunitiesHostApp(MY_ROWS) + async with app.run_test() as pilot: + screen = app.screen_under_test + screen.query_one("#my_communities_table", DataTable).move_cursor(row=1) + screen.action_leave_community() + await pilot.pause() + assert app.screen is screen + app.leave_community.assert_not_called() + + @pytest.mark.asyncio + async def test_leave_with_no_communities(self): + app = CommunitiesHostApp([]) + async with app.run_test() as pilot: + app.screen_under_test.action_leave_community() + await pilot.pause() + assert app.screen is app.screen_under_test + + @pytest.mark.asyncio + async def test_quit(self): + app = CommunitiesHostApp([]) + async with app.run_test() as pilot: + app.screen_under_test.action_quit() + await pilot.pause() + assert app.dismissed == 200 + + +async def _open_community(app, pilot) -> CommunityScreen: + app.screen_under_test.action_open_community() + await pilot.pause() + assert isinstance(app.screen, CommunityScreen) + return app.screen + + +class TestCommunityScreen: + @pytest.mark.asyncio + async def test_shows_members_posts_and_comments(self): + app = CommunitiesHostApp(MY_ROWS) + async with app.run_test() as pilot: + screen = await _open_community(app, pilot) + app.get_community_blog_posts.assert_called_once_with("ds-guid") + app.get_community_comments.assert_called_once_with("ds-guid") + assert screen.query_one("#community_members_table", DataTable).row_count == 1 + posts = screen.query_one("#community_posts_table", DataTable) + assert [posts.get_row_at(i)[0] for i in range(posts.row_count)] == ["Second", "First"] + assert screen.query_one("#community_comments_table", DataTable).row_count == 1 + # the highlighted post's full text is shown below the list + assert "Again" in str(screen.query_one("#community_post_text", Static).render()) + + @pytest.mark.asyncio + async def test_highlighting_a_post_shows_its_text(self): + app = CommunitiesHostApp(MY_ROWS) + async with app.run_test() as pilot: + screen = await _open_community(app, pilot) + screen.query_one("#community_posts_table", DataTable).move_cursor(row=1) + await pilot.pause() + assert "Hello" in str(screen.query_one("#community_post_text", Static).render()) + + @pytest.mark.asyncio + async def test_empty_blog(self): + app = CommunitiesHostApp(MY_ROWS) + app.posts = [] + async with app.run_test() as pilot: + screen = await _open_community(app, pilot) + assert "No posts yet" in str(screen.query_one("#community_post_text", Static).render()) + + @pytest.mark.asyncio + async def test_new_blog_post(self): + app = CommunitiesHostApp(MY_ROWS) + async with app.run_test() as pilot: + screen = await _open_community(app, pilot) + await pilot.press("ctrl+n") + await pilot.pause() + assert isinstance(app.screen, NewBlogPostScreen) + app.screen.query_one("#blog_post_title_input", Input).value = "Hi all" + app.screen.query_one("#blog_post_text", TextArea).text = "First post here" + await pilot.press("ctrl+s") + await pilot.pause() + app.add_community_blog_post.assert_called_once_with("ds-guid", "Data Science SIG", "Hi all", + "First post here") + assert app.screen is screen + assert screen.query_one("#community_posts_table", DataTable).get_row_at(0)[0] == "Hi all" + + @pytest.mark.asyncio + async def test_blog_post_needs_title_and_text(self): + app = CommunitiesHostApp(MY_ROWS) + async with app.run_test() as pilot: + await _open_community(app, pilot) + app.screen.action_new_post() + await pilot.pause() + app.screen.query_one("#blog_post_title_input", Input).value = "No text" + app.screen.action_submit() + await pilot.pause() + assert isinstance(app.screen, NewBlogPostScreen) + app.add_community_blog_post.assert_not_called() + + @pytest.mark.asyncio + async def test_cancel_blog_post(self): + app = CommunitiesHostApp(MY_ROWS) + async with app.run_test() as pilot: + screen = await _open_community(app, pilot) + screen.query_one("#new_blog_post_btn", Button).press() + await pilot.pause() + await pilot.press("escape") + await pilot.pause() + assert app.screen is screen + app.add_community_blog_post.assert_not_called() + + @pytest.mark.asyncio + async def test_add_comment_from_the_comments_list(self): + app = CommunitiesHostApp(MY_ROWS) + async with app.run_test() as pilot: + screen = await _open_community(app, pilot) + screen.query_one("#community_comments_table", DataTable).focus() + await pilot.press("ctrl+n") + await pilot.pause() + assert isinstance(app.screen, AddCommunityCommentScreen) + app.screen.query_one("#community_comment_type", Select).value = "SUGGESTION" + app.screen.query_one("#community_comment_text", TextArea).text = "Meet monthly?" + app.screen.query_one("#submit_community_comment_btn", Button).press() + await pilot.pause() + app.add_community_comment.assert_called_once_with("ds-guid", "Meet monthly?", "SUGGESTION") + assert app.screen is screen + assert screen.query_one("#community_comments_table", DataTable).row_count == 2 + + @pytest.mark.asyncio + async def test_comment_needs_text(self): + app = CommunitiesHostApp(MY_ROWS) + async with app.run_test() as pilot: + await _open_community(app, pilot) + app.screen.action_add_comment() + await pilot.pause() + app.screen.action_submit() + await pilot.pause() + assert isinstance(app.screen, AddCommunityCommentScreen) + app.add_community_comment.assert_not_called() + + @pytest.mark.asyncio + async def test_failed_post_leaves_the_list_unchanged(self): + app = CommunitiesHostApp(MY_ROWS) + app.add_community_blog_post = MagicMock(return_value=False) + async with app.run_test() as pilot: + screen = await _open_community(app, pilot) + app.screen.action_new_post() + await pilot.pause() + app.screen.query_one("#blog_post_title_input", Input).value = "Hi" + app.screen.query_one("#blog_post_text", TextArea).text = "Text" + app.screen.action_submit() + await pilot.pause() + assert app.get_community_blog_posts.call_count == 1 + assert screen.query_one("#community_posts_table", DataTable).row_count == 2 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 index 238f27ca..4e534515 100644 --- a/tests/micro-tests/my_profile/test_connection_and_change_user.py +++ b/tests/micro-tests/my_profile/test_connection_and_change_user.py @@ -15,12 +15,15 @@ from ConnectionErrorScreen import ConnectionErrorScreen, RETRY_INTERVAL from SplashScreen import SplashScreen from profile_utils import check_egeria_access -from pyegeria import PyegeriaConnectionException +import httpx +from pyegeria import (PyegeriaConnectionException, PyegeriaTimeoutException, PyegeriaUnauthorizedException, + PyegeriaClientException) -BAD_CREDENTIALS_BODY = ( - '{"class":"FFDCResponseBase","relatedHTTPCode":401,' - '"exceptionErrorMessage":"User authentication failed for user: garygeeke - Bad credentials"}' -) +TOKEN_URL = "https://localhost:9443/api/token" + + +def _token_response(status: int, text: str) -> httpx.Response: + return httpx.Response(status, text=text, request=httpx.Request("POST", TOKEN_URL)) class HostApp(App): @@ -57,10 +60,28 @@ def test_unreachable_platform(self, mock_egeria): 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 + def test_token_request_timeout_counts_as_unreachable(self, mock_egeria): + mock_egeria.return_value.create_egeria_bearer_token.side_effect = PyegeriaTimeoutException( + None, None, httpx.ReadTimeout("timed out")) + 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." + mock_egeria.return_value.close_session.assert_called_once() + + @patch("pyegeria.Egeria") + def test_bad_credentials(self, mock_egeria): + mock_egeria.return_value.create_egeria_bearer_token.side_effect = PyegeriaUnauthorizedException( + _token_response(401, "Unauthorized: invalid user id or password"), None, {"userid": "garygeeke"}) error = check_egeria_access("qs-view-server", "https://localhost:9443", "garygeeke", "wrong") - assert "Bad credentials" in error + assert error == "Egeria rejected the credentials for user garygeeke." + mock_egeria.return_value.close_session.assert_called_once() + + @patch("pyegeria.Egeria") + def test_other_error_is_one_line(self, mock_egeria): + mock_egeria.return_value.create_egeria_bearer_token.side_effect = PyegeriaClientException( + _token_response(500, "boom"), None, {"userid": "garygeeke"}) + error = check_egeria_access("qs-view-server", "https://localhost:9443", "garygeeke", "secret") + assert error.startswith("Egeria at https://localhost:9443 returned an error: ") + assert "\n" not in error class TestConnectionErrorScreen: diff --git a/tests/micro-tests/my_profile/test_key_bindings.py b/tests/micro-tests/my_profile/test_key_bindings.py index a4409248..0c0b923d 100644 --- a/tests/micro-tests/my_profile/test_key_bindings.py +++ b/tests/micro-tests/my_profile/test_key_bindings.py @@ -19,6 +19,8 @@ import AddCommentScreen import AddToElementsScreens +import CommunitiesScreen +import CommunityScreen import CreateProfileScreen import CreateSubscriptionRequestScreen import EditElementsScreens @@ -40,7 +42,7 @@ from my_egeria.DemoCode.My_Profile.my_egeria import MyProfileApp MODULES = [ - AddCommentScreen, AddToElementsScreens, CreateProfileScreen, CreateSubscriptionRequestScreen, + AddCommentScreen, AddToElementsScreens, CommunitiesScreen, CommunityScreen, CreateProfileScreen, CreateSubscriptionRequestScreen, EditElementsScreens, FeedbackScreens, GenericDataViewScreen, MainScreen, MyBookMarksScreen, MyNetworkScreen, MyTeamScreen, SearchForTermScreen, SelectionOverviewScreen, ShopForDataScreen, ShowCommentsScreen, SplashScreen, StatusScreen, TechnologyTypeScreens, UserIdentitiesScreen, diff --git a/tests/micro-tests/my_profile/test_my_egeria.py b/tests/micro-tests/my_profile/test_my_egeria.py index effde34f..28dbfe2a 100644 --- a/tests/micro-tests/my_profile/test_my_egeria.py +++ b/tests/micro-tests/my_profile/test_my_egeria.py @@ -273,6 +273,15 @@ async def test_my_network_option_shows_network(self): await app.handle_option_selected(event) app.show_my_network.assert_called_once_with() + @pytest.mark.asyncio + async def test_work_with_communities_option_shows_communities(self): + app = MyProfileApp() + app.show_communities = MagicMock() + event = MagicMock() + event.option.prompt = "[] Work with Communities" + await app.handle_option_selected(event) + app.show_communities.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()