diff --git a/.github/workflows/test-publish.yaml b/.github/workflows/test-publish.yaml new file mode 100644 index 0000000..3acb26f --- /dev/null +++ b/.github/workflows/test-publish.yaml @@ -0,0 +1,55 @@ +name: Release and Publish + +on: + pull_request: + types: [closed] + branches: + - master + +jobs: + release-and-publish: + if: github.event.pull_request.merged == true + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - name: Set up Python 3.9 + uses: actions/setup-python@v4 + with: + python-version: "3.9" + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install --upgrade pip setuptools wheel twine build + pip install packaging==22 + + - name: Build package + run: | + rm -rf dist/ build/ *.egg-info + python setup.py sdist bdist_wheel + twine check dist/* + + - name: Get version + id: get_version + run: | + version=$(python setup.py --version) + echo "version=${version}" >> $GITHUB_OUTPUT + + - name: Create Release + uses: softprops/action-gh-release@v1 + with: + tag_name: v${{ steps.get_version.outputs.version }} + name: Release v${{ steps.get_version.outputs.version }} + body: | + Release for PR #${{ github.event.pull_request.number }} + ${{ github.event.pull_request.title }} + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + + - name: Publish to TestPyPI + env: + TWINE_USERNAME: __token__ + TWINE_PASSWORD: ${{ secrets.TEST_PYPI_API_TOKEN }} + run: | + twine upload --repository testpypi dist/* --verbose diff --git a/.gitignore b/.gitignore index 1c97479..5e6da28 100644 --- a/.gitignore +++ b/.gitignore @@ -22,4 +22,5 @@ zenduty/.DS_Store __pycache__/ *.pyc venv/ -.idea/ +assets/ +/.pytest_cache/ diff --git a/README.md b/README.md index 5c0ef2d..ef28b8b 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,7 @@ +# What is Zenduty?? + +Zenduty is a cutting edge platform for incident management. With high level automation, Zenduty enables faster and better incident resolution keeping developers first. + # Zenduty Python SDK Python SDK to communicate with zenduty endpoints @@ -5,46 +9,1185 @@ Python SDK to communicate with zenduty endpoints ## Installing Installation can be done through pip, as follows: -``` + +```sh $ pip install zenduty-api ``` + or you may grab the latest source code from GitHub: -``` + +```sh $ git clone https://github.com/Zenduty/zenduty-python-sdk -$ python3 setup.py install ``` + ## Contents -1) zenduty/api : contains the functions to communicate with zenduty API endpoints -2) zenduty/ : contains the common required files -3) bin/ : contains sample script to run zenduty functions + +1. zenduty/apiV2 : Contains proper Clients to communicate and execute each function +2. zenduty/api : contains the functions to communicate with zenduty API endpoints +3. zenduty/ : contains the common required files +4. bin/ : contains sample script to run zenduty functions ## Getting started Before you begin making use of the SDK, make sure you have your Zenduty Access Token. - You can then import the package into your python script. + +First of all, start off by making a client which connects to Zenduty using API Token. And create a team, most of the operations we'd do start off by creating a team, and creating services. For now, we will start off with creating an instance of a team. + +The Approach here is to make clients here, every module will get a new client to make things simpler and easier for us to understand. + +You will also notice most of the create functions will be creating an object, this is because we will need to pass them to update, and delete functions later on. you are free to use the sdk as you like. + +```python + +from zenduty.apiV2.authentication.zenduty_credential import ZendutyCredential +from zenduty.apiV2.client import ZendutyClient + +class SDKTestingClient: + def __init__(self): + self.cred = ZendutyCredential() + self.client = ZendutyClient( + credential=self.cred, use_https=True + ) # defaults to default service endpoint zenduty.com + self.datetime_timestamp = self.datetime_timestamp() + + @staticmethod + def datetime_timestamp(): + return datetime.now().strftime("%Y-%m-%d %H:%M:%S") +``` + +It is important to note that each function returns a urllib3.response.HTTPResponse object. + +## Teams + +This object represents a team of the account. It lets you create different independent operational units in the account. You can check out the team docs here https://docs.zenduty.com/docs/teams. + +A Team can have multiple Members, Services, Integrations, Schedules, Escalation Policies, Priorities, Maintenance, etc.. + +```python +class SDKTeamsClient(SDKTestingClient): + def __init__(self): + super().__init__() + self.teams_client = TeamsClient(client=self.client) + self.invite_url = "https://zenduty.com/api/invite/accept/" + self.test_team_name = f"Team - {self.datetime_timestamp}" ``` -import zenduty + +#### POST - Create a new team + +```python +def create_team(self): + create_team = self.teams_client.create_team(self.test_team_name) + self.team_obj = create_team # class object made to use later down the road + return create_team ``` -Based on the endpoint you want to communicate with, create an object of the required class. For example, to create an incident: + +From here on, use this team object returned by create_team() to get the unique id of that team by calling it like this. + +#### GET - List Teams + +Will fetch all the teams present in that account + +```python +def list_teams(self): + list_teams = self.teams_client.list_teams() + return list_teams ``` -api_obj = zenduty.IncidentsApi(zenduty.ApiClient('your-access-token')) -body = {"service":"c7fff4c5-2def-41e8-9120-c63f649a825c", - "escalation_policy":"a70244c8-e343-4dd0-8d87-2f767115568a", - "user":null, - "title":"Name of trial", - "summary":"summary of trial"} -response = api_obj.create_incident(body) -print(response.data) -print(response.status_code) + +#### PATCH - Update teams + +Update the team + +```python +def update_team(self): + update_team = self.teams_client.update_team(self.team_obj, name="Updated team name here") + return update_team +``` + +#### DEL - Delete team + +Delete the team + +```python +def delete_team(self): + return self.teams_client.delete_team(self.team_obj) +``` + +## Account Member + +This object represents an account user. Each account member object has a role, which can be "owner," "admin," or "user." An account can have only one owner, but multiple admins and users. + +Prerequisite: A team must be created, where the role of each member can be assigned. For our example, we are creating a new team using the create_team() method mentioned above. + +```python + +class SDKAccountMembersClient(SDKTeamsClient): + def __init__(self): + super().__init__() + self.teams_client = TeamsClient(client=self.client) + self.account_member_client = AccountMemberClient(client=self.client) + self.team_obj = self.create_team() #Will be used for the purpose of inviting users to the team + +``` + +#### GET - Invite a member to the team + +#### Invite a member to the team. + +```python +def test_invite(self): + test_email = f"john.doe.{random.randint(2,10000000000000000000000)}@zenduty.com" + self.team_invite_object = self.account_member_client.invite( + team_id=self.team_obj.unique_id, + email=test_email, + first_name="Test", + last_name="User", + role=2, + ) + +``` + +#### PATCH - Update Account Member + +```python +def test_update_account_member(self): + result = self.account_member_client.update_account_member( + self.team_invite_object, + "Updated", + "Member Details", + 2, + ) +``` + +#### GET - Get Account member + +#### Get details about a particular team member (NOTE: To get account member you have to pass its user id, not AccountMember Uniqueid) + +```python +def test_get_account_member_by_id(self): + result = self.account_member_client.get_account_member(self.team_invite_object.user.username) +``` + +#### GET - Get all the members of a team + +#### Get details of all the members of the team. + +```python +def test_get_all_members(self): + result = self.account_member_client.get_all_members() +``` + +#### DEL - Delete an Account member + +#### Delete a particular member of the team. + +```python +def test_delete_account_member(self): + self.account_member_client.delete_account_member(self.team_invite_object) +``` + +## Account Roles + +There are a list of permissions you could give to a role. Please refer to these docs, https://apidocs.zenduty.com/#tag/Account-Custom-Role. + +```python +class SDKAccountRolesClient(SDKTestingClient): + def __init__(self): + super().__init__() + self.account_role_client = AccountRoleClient(client=self.client) +``` + +#### POST - Create Account Role + +```python +def create_account_role(self): + test_name = f"Account Role - {self.datetime_timestamp}" + create_account_role = self.account_role_client.create_account_role( + name=test_name, + description="Account Role Description", + permissions=["sla_read"], + ) +``` + +#### GET - Get an Account Role + +```python +def test_get_account_role(self): + get_account_role = self.account_role_client.get_account_role(self.role_obj.unique_id) + return get_account_role +``` + +#### GET - Get a list of roles + +```python +def test_list_account_roles(self): + list_account_roles = self.account_role_client.list_account_roles() + return list_account_roles +``` + +#### PATCH - Update an Account Role + +```python + +def test_update_account_role(self): + update_account_role = self.account_role_client.update_account_role(account_role=self.role_obj, permissions=["sla_read"]) + return update_account_role + +``` + +#### DEL - Delete an Account Role + +```python +def test_delete_account_role(self): + delete_account_role = self.account_role_client.delete_account_role(account_role=self.role_obj) +``` + +## Global Event Router + +Global Event Router is a webhook, when sent requests to it, would navigate it to a particular integration, to a particular request, if matched with the alert rules defined, would raise an alert. + +Refer to this, for more information, https://apidocs.zenduty.com/#tag/Global-Router. + +```python +class SDKGERClients(SDKTestingClient): + def __init__(self): + super().__init__() + self.router_client = RouterClient(client=self.client) + self.router_name = f"Router - {self.datetime_timestamp}" +``` + +#### POST - Create Router + +```python +def create_router(self): + create_router = self.router_client.create_router( + name=self.router_name, + description="Router Description", + ) +``` + +#### GET - List Routers + +```python +def get_all_routers(self): + return self.router_client.get_all_routers() +``` + +#### GET - Get Router by ID + +```python +def get_router_by_id(self): + return self.router_client.get_router_by_id(self.router_obj.unique_id) +``` + +#### PATCH - Update a particular Router + +```python +def update_router(self): + self.router_client.update_router( + router = self.router_obj, + name = f"Router - {self.datetime_timestamp}", + description = "New Router Description", + ) +``` + +#### DEL - Delete a particular Router + +```python +def delete_router(self): + return self.router_client.delete_router(self.router_obj) +``` + +## Events + +This object represents the events of an integration. + +```python +class SDKEventsClient(SDKTestingClient): + def __init__(self): + super().__init__() + self.event_client = EventClient(client=self.client) + self.event_name = f"Event - {self.datetime_timestamp}" +``` + +#### GET - Get Router Client + +```python +def get_router_client(self): + get_router = self.event_client.get_router_client() + +``` + +#### POST - Create an Event + +```python +def test_create_event(self): + create_event = self.event_client.create_event( + integration_key=, + alert_type="info", + message="This is info alert", + summary="This is the incident summary111", + entity_id=123455, + payload={ + "status": "ACME Payments are failing", + "severity": "1", + "project": "kubeprod", + }, + urls=[ + { + "link_url": "https://www.example.com/alerts/12345/", + "link_text": "Alert URL", + } + ], + ) +``` + +## Escalation Policy + +Escalation policies dictate how an incident created within a service escalates within your team. + +```python + +class SDKEscalationPolicyClient(SDKTeamsClient): + # Inheriting a few methods from the Teams Object. + def __init__(self): + super().__init__() + self.uuid = self.generate_uuid() + self.teams_client = TeamsClient(client=self.client) + self.account_member_client = AccountMemberClient(client=self.client) + + # Team object: Needed for escalation policy client + self.team_obj = self.teams_client.create_team(name=f"ESP Team - {self.datetime_timestamp}") + + self.escalation_policy_client = self.teams_client.get_escalation_policy_client( + self.team_obj + ) + + self.ep_name = f"EP - {self.datetime_timestamp}" + + # User required to add to escalation policy + self.test_email = f"john.doe.{random.randint(2,10000000000000000000000)}@zenduty.com" + self.team_member_obj = self.account_member_client.invite( + team_id=self.team_obj.unique_id, + email=self.test_email, + first_name="Test", + last_name="User", + role=2, + ) + + def generate_uuid(self): + return uuid4() +``` + +#### POST - Create an Escalation Policy + +```python +def create_escalation_policy(self): + self.rule_build = [ + { + "delay": 0, + "targets": [ + {"target_type": 2, "target_id": self.team_member_obj.user.username} + ], + "position": 1, + } + ] + + self.create_escalation_policy_obj = self.escalation_policy_client.create_esp( + self.ep_name, rules=self.rule_build + ) + +``` + +#### GET - Get Escalation Policies by ID + +```python +def test_get_esp_by_id(self): + self.escalation_policy_client.get_esp_by_id( + esp_id=self.create_escalation_policy_obj.unique_id + ) +``` + +#### POST - Update Escalation Policy + +```python +def test_update_esp(self): + update_esp = self.escalation_policy_client.update_esp( + esp=self.create_escalation_policy_obj, + name="Test Updated", + rules=self.rule_build, + ) +``` + +#### GET - Get all the escalation policies + +```python +def test_get_all_policies(self): + all_esp = self.escalation_policy_client.get_all_policies() + return all_esp +``` + +#### DEL - Delete an Escalation Policy + +```python +def test_delete_esp(self): + delete_esp = self.escalation_policy_client.delete_esp(esp=self.create_escalation_policy_obj) + +``` + +## Schedules + +```python +class SDKSchedulesClient(SDKTeamsClient): + def __init__(self): + super().__init__() + self.uuid = self.generate_uuid() + self.teams_client = TeamsClient(client=self.client) + self.account_member_client = AccountMemberClient(client=self.client) + + self.team_obj = self.teams_client.create_team(name=f"Schedule Team - {self.datetime_timestamp}") + + # adding test user to the team + self.test_email = f"john.doe.{random.randint(2,10000000000000000000000)}@zenduty.com" + self.team_member_obj = self.account_member_client.invite( + team_id=self.team_obj.unique_id, + email=self.test_email, + first_name="Test", + last_name="User", + role=2, + ) + + self.schedules_client = self.teams_client.get_schedule_client(self.team_obj) + self.schedules_name = f"Schedules - {self.datetime_timestamp}" + self.layers = [ + { + "name": "Layer 1", + "is_active": True, + "restriction_type": 0, + "restrictions": [], + "rotation_start_time": "2025-07-29T03:30:00.000Z", + "rotation_end_time": None, + "shift_length": 86400, + "users": [ + { + "user":self.team_member_obj.user.username, + "position": 1, + } + ], + } + ] + + def generate_uuid(self): + return uuid4() +``` + +#### POST - Create an Escalation Policy + +```python +def create_schedule(self): + create_schedule = self.schedules_client.create_schedule( + name=self.schedules_name, + timezone="Asia/Kolkata", + layers=self.layers, + overrides=self.overrides, + ) +``` + +#### GET - Get all Schedules + +```python +def test_get_all_schedules(self): + get_all_schedules = self.schedules_client.get_all_schedules() +``` + +#### GET - Get Schedules by ID + +```python +def test_get_schedule_by_id(self): + self.get_schedule_by_id = self.schedules_client.get_schedule_by_id( + schedule_id=self.schedule_obj.unique_id + ) + return self.get_schedule_by_id +``` + +#### POST - Update a Schedule + +```python +def test_update_schedule(self): + update_schedule = self.schedules_client.update_schedule( + schedule=self.schedule_obj, + name="Test Schedule Updated", + ) +``` + +#### DEL - Delete a Schedule + +```python +def test_delete_schedules(self): + delete_schedule = self.schedules_client.delete_schedule( + schedule=self.schedule_obj + ) +``` + +## Maintenance + +```python +class SDKMaintenanceClient(SDKTeamsClient): + def __init__(self): + super().__init__() + self.uuid = self.generate_uuid() + self.teams_client = TeamsClient(client=self.client) + + + self.team_by_id = self.teams_client.find_team_by_id( + team_id=self.team_ids[0].unique_id + ) + self.maintenance_client = self.teams_client.get_maintenance_client( + self.team_by_id + ) + self.maintenance_name = f"Maintenance Mode - {self.datetime_timestamp}" + + def generate_uuid(self): + return uuid4() +``` + +#### POST - Create a Maintenance + +```python +def test_create_maintenance(self): + self.maintenance_obj = self.maintenance_client.create_team_maintenance( + name=self.maintenance_name, + start_time="2026-07-08T18:06:00", + end_time="2026-07-08T18:06:00", + service_ids=[], + ) +``` + +#### GET - Get all Maintenances + +```python +def test_get_maintenance_by_id(self): + get_maintenance_by_id = self.maintenance_client.get_maintenance_by_id( + maintenance_id=self.maintenance_obj.unique_id + ) +``` + +#### PATCH - Update a Maintenance + +```python +def test_update_maintenance(self): + update_maintenance = self.maintenance_client.update_maintenance( + maintenance_id=self.maintenance_obj, + name="Updated Maintenance Name", + start_time="2026-07-08T18:06:00", + end_time="2026-07-08T18:06:00", + service_ids=[], + ) +``` + +#### DEL - Delete a Maintenance + +```python +def test_delete_maintenance(self): + delete_maintenance = self.maintenance_client.delete_maintenance( + maintenance_id=self.maintenance_obj + ) +``` + +## Incidents + +What is an Incident?? + +An incident on Zenduty is an event that is not part of usual operations, and that disrupts operational processes within a Service that is owned by a team. Incidents can be automatically created by an alert integration within the service or manually by a user. + +An incident on Zenduty has three states: + +Triggered: Triggered is the first state of the incident. Zenduty will continue escalating the alert, depending on the escalation policy, as long as the incident is in the Triggered state. +Acknowledged: When an incident is acknowledged by a user, Zenduty stops all further escalations. +Resolved: Marking an incident as resolved implies that the incident has been remediated. Incidents can be resolved automatically by the service integration that created it, or manually by a user. + +```python +class SDKIncidentsClient(SDKTestingClient): + def __init__(self): + super().__init__() + self.incident_client = IncidentClient(client=self.client) + self.incident_name = f"Incident - {self.datetime_timestamp}" + self.incident_notes = f"Incident Notes - {self.datetime_timestamp}" + self.incident_tags = f"Incident Tags - {self.datetime_timestamp}" + + self.service_client = SDKServicesClient() + + + def generate_uuid(self): + return uuid4() +``` + +#### POST - Create an Incident + +```python +def create_incident(self): + create_incident = self.incident_client.create_incident( + title=self.incident_name, service= + ) +``` + +#### POST - Create an Incident Note + +NOTE: Here we are creating a note client using incident object, meaning it is important to first execute create incident before running this function. + +```python +def test_create_note(self): + # to create a notes functions + self.note_client = self.incident_client.get_note_client( + incident=self.incident_obj + ) + + self.incident_note_obj = self.note_client.create_incident_note( + note=self.incident_notes + ) +``` + +#### GET - Get all Incident Notes + +```python +def test_get_all_incident_notes(self): + get_all_incident_notes = self.note_client.get_all_incident_notes() +``` + +#### GET - Get Incident note by id + +```python +def test_get_incident_by_note_id(self): + get_incident_note_by_id = self.note_client.get_incident_note_by_id( + incident_note_unique_id=self.incident_note_obj.unique_id + ) +``` + +#### PATCH - Update an Incident note + +```python +def test_update_incident_note(self): + update_incident_note = self.note_client.update_incident_note( + incident_note=self.incident_note_obj, + note="Updated Incident Note", + ) +``` + +#### DEL - Delete an Incident note + +```python +def test_delete_incident_note(self): + delete_incident_note = self.note_client.delete_incident_note( + incident_note=self.incident_note_obj + ) +``` + +#### POST - Create an Incident Tag + +NOTE: All the functions below require a tag client to be made, There we must run this function before executing any incident tags functions. + +```python +def test_create_tag_client(self): + self.tag_client = self.incident_client.get_tags_client(self.incident_obj) +``` + +#### GET - Get all Incident Tags + +```python +def test_get_all_tags(self): + get_all_tags = self.tag_client.get_all_tags() + return get_all_tags +``` + +#### GET - Get all Incidents + +```python +def test_get_all_incidents(self): + get_all_incidents = self.incident_client.get_all_incidents(page=1) + return get_all_incidents +``` + +#### GET - Get Alerts of Incidents + +```python +def test_get_alert_on_incidents(self): + get_alerts_by_incident = self.incident_client.get_alerts_for_incident( + incident_number=self.incident_obj.incident_number + ) +``` + +#### PATCH - Update an Incident + +```python +def test_update_incident(self): + update_incident = self.incident_client.update_incident( + incident_id=self.incident_obj.unique_id, + title="Updated Incident Name", + status=3, + service="a91a3a00-8de9-472c-ad2e-61e7c89db062", + ) +``` + +## Postmortem + +```python +class SDKPostMortemClient(SDKTeamsClient): + def __init__(self): + super().__init__() + self.uuid = self.generate_uuid() + self.teams_client = TeamsClient(client=self.client) + self.team_obj = self.teams_client.create_team(name=f"Post Mortem Team - {self.datetime_timestamp}") + + + self.incident_client = IncidentClient(client=self.client) + self.incident_name = f"Incident - {self.datetime_timestamp}" + + self.postmortem_client = self.teams_client.get_postmortem_client( + self.team_obj + ) + + self.postmortem_name = f"Postmortem - {self.datetime_timestamp}" + + def generate_uuid(self): + return uuid4() +``` + +#### POST - Create a Postmortem + +```python +def test_create_postmortem(self): + # create new service + self.service_client = SDKServicesClient() + self.service_obj = self.service_client.test_create_service() + + # Create the Incident + create_incident = self.incident_client.create_incident( + title=self.incident_name, service=self.service_obj.unique_id + ) + + # Create the Postmortem + self.postmortem_obj= self.postmortem_client.create_postmortem( + author="43b2493a-58e9-4454-9fe5-4", + incidents=[create_incident.unique_id], + title="Test Postmortem", + ) +``` + +#### GET - Get postmortem by id + +```python +def test_get_postmortem_by_id(self): + self.postmortem_by_id = self.postmortem_client.get_postmortem_by_id( + postmortem_id=self.postmortem_obj.unique_id + ) +``` + +#### PATCH - Update a postmortem + +```python +def test_update_postmortem(self): + update_postmortem = self.postmortem_client.update_postmortem( + self.postmortem_obj, + author="43b2493a-58e9-4454-9fe5-4", + incidents=[], + title="Test Postmortem Updated", + ) + +``` + +#### DEL - Delete a postmortem + +```python +def test_delete_postmortem(self): + delete_postmortem = self.postmortem_client.delete_postmortem( + self.postmortem_obj + ) +``` + +## Priorities + +```python +class SDKPrioritiesClient(SDKTeamsClient): + def __init__(self): + super().__init__() + self.uuid = self.generate_uuid() + self.teams_client = TeamsClient(client=self.client) + + self.team_obj = self.teams_client.create_team(name=f"Post Mortem Team - {self.datetime_timestamp}") + + self.priority_client = self.teams_client.get_priority_client(self.team_obj) + self.priority_name = f"Priority - {self.datetime_timestamp}" + + def generate_uuid(self): + return uuid4() +``` + +#### POST - Create a priority + +```python +def test_create_priority(self): + self.priority_obj = self.priority_client.create_priority( + name=self.priority_name, + description="Priority Description", + color="red", + ) +``` + +#### GET - Get all priorities + +```python +def test_get_all_priorities(self): + get_all_priorities = self.priority_client.get_all_priorities() + return get_all_priorities +``` + +#### GET - Get priorities by ID + +```python +def test_get_priority_by_id(self): + self.priority_by_id = self.priority_client.get_priority_by_id( + priority_id=self.priority_obj.unique_id + ) +``` + +#### PATCH - Update the priority + +```python +def test_update_priority(self): + update_priority = self.priority_client.update_priority( + self.priority_obj, + name="Test Priority Updated", + description="Test Priority", + ) +``` + +#### DEL - Delete a priority + +```python +def test_delete_priority(self): + delete_priority = self.priority_client.delete_priority(self.priority_obj) +``` + +## Roles + +```python +class SDKRolesClient(SDKTeamsClient): + def __init__(self): + super().__init__() + self.uuid = self.generate_uuid() + self.teams_client = TeamsClient(client=self.client) + self.team_obj = self.teams_client.create_team(name=f"Roles Team - {self.datetime_timestamp}") + + self.role_client = self.teams_client.get_incident_role_client(self.team_obj) + self.role_name = f"Role - {self.datetime_timestamp}" + + def generate_uuid(self): + return uuid4() +``` + +#### POST - Create a Role + +```python +def test_create_role(self): + self.role_obj = self.role_client.create_incident_role( + title="Test Role", + description="Test Role", + rank=2, + ) +``` + +#### GET - Get incident role by id + +```python +def test_get_role_by_id(self): + self.get_role_by_id = self.role_client.get_incident_role_by_id( + role_id=self.role_obj.unique_id + ) +``` + +#### PATCH - Update an incident role + +```python +def test_update_role(self): + self.update_role = self.role_client.update_incident_role( + role=self.role_obj, + title="Test Role Updated", + ) +``` + +#### DEL - Delete an incident role + +```python +def test_delete_role(self): + self.delete_role = self.role_client.delete_incident_role( + role=self.role_obj + ) +``` + +## Services + +```python +class SDKServicesClient(SDKTeamsClient): + def __init__(self): + super().__init__() + # Making the Teams Client + self.teams_client = TeamsClient(client=self.client) + self.team_obj = self.teams_client.create_team(name=f"Services Team - {self.datetime_timestamp}") + + + # Making the Service Client + self.service_client = self.teams_client.get_service_client(self.team_obj) + + self.escalation_policy_client = self.teams_client.get_escalation_policy_client( + self.team_obj + ) + self.priority_client = self.teams_client.get_priority_client(self.team_obj) + self.sla_client = self.teams_client.get_sla_client(self.team_obj) + + # Making the names + self.ep_name = f"EP - {self.datetime_timestamp}" + self.priority_name = f"Priority - {self.datetime_timestamp}" + self.sla_name = f"SLA - {self.datetime_timestamp}" + self.service_name = f"Service - {self.datetime_timestamp}" + + + self.account_member_client = AccountMemberClient(client=self.client) + self.test_email = f"john.doe.{random.randint(2,10000000000000000000000)}@zenduty.com" + self.team_member_obj = self.account_member_client.invite( + team_id=self.team_obj.unique_id, + email=self.test_email, + first_name="Test", + last_name="User", + role=2, + ) +``` + +#### POST - Create a Service + +```python +def test_create_service(self): + # Create the escalation policy + self.rule_build = [ + { + "delay": 0, + "targets": [ + {"target_type": 2, "target_id": self.team_member_obj.user.username} + ], + "position": 1, + } + ] + create_escalation_policy = self.escalation_policy_client.create_esp( + self.ep_name, rules=self.rule_build + ) + + + # Create the Priority + create_priority = self.priority_client.create_priority( + name=self.priority_name, + description="Priority Description", + color="red", + ) + + + + # Create the SLA + create_sla = self.sla_client.create_sla(name="Test SLA", escalations=[]) + + # Finally create the service + create_service = self.service_client.create_service( + name=f"Test Service - {self.datetime_timestamp}", + escalation_policy=create_escalation_policy.unique_id, + team_priority=create_priority.unique_id, + sla=create_sla.unique_id, + ) + + return create_service +``` + +## Integrations + +#### POST - Create an integration + +```python +integration_client = service_client.get_integration_client(svc=) + +create_integration = integration_client.create_intg( name="Test Integration", summary="Test Integration", application=) + +``` + +#### GET - Get all integrations + +```python +all_integrations = integration_client.get_all_integrations() +``` + +#### GET - Get integration by id + +```python +integration_by_id = integration_client.get_intg_by_id(intg=) +``` + +#### PATCH - Update an integration + +```python +update_integration = integration_client.update_intg(intg=, name="Test Integration Updated", application=) +``` + +#### DEL - Delete an integration + +```python +delete_integration = integration_client.delete_intg(intg=) +``` + +## SLA + +```python +class SDKSLAClient(SDKTeamsClient): + def __init__(self): + super().__init__() + self.teams_client = TeamsClient(client=self.client) + self.account_member_client = AccountMemberClient(client=self.client) + self.team_obj = self.teams_client.create_team(name=f"Integration Team - {self.datetime_timestamp}") + + self.sla_client = self.teams_client.get_sla_client(self.team_obj) +``` + +#### POST - Create an SLA + +```python +def test_create_sla(self): + self.sla_obj = self.sla_client.create_sla(name="Test SLA", escalations=[]) +``` + +#### GET - Get SLA by id + +```python +def test_get_sla_by_id(self): + sla_by_id = self.sla_client.get_sla_by_id(sla_id=self.sla_obj.unique_id) +``` + +#### PATCH - Update SLA + +```python +def test_update_sla(self): + update_sla = self.sla_client.update_sla(sla=self.sla_obj, name="Test SLA Updated", escalations=[]) +``` + +#### DEL - Delete SLA + +```python +def test_delete_sla(self): + delete_sla = self.sla_client.delete_sla(sla=self.sla_obj) +``` + +## Tags + +```python +class SDKTagClient(SDKTeamsClient): + def __init__(self): + super().__init__() + self.teams_client = TeamsClient(client=self.client) + self.account_member_client = AccountMemberClient(client=self.client) + self.team_obj = self.teams_client.create_team(name=f"Integration Team - {self.datetime_timestamp}") + + self.tag_client = self.teams_client.get_tag_client(self.team_obj) +``` + +#### POST - Create a tag + +```python +def test_create_tag(self): + self.tag_obj = self.tag_client.create_tag(name=f"Tag name - {self.datetime_timestamp}", color="red") +``` + +#### GET - Get all tags + +```python +def test_get_all_tags(self): + get_all_tags = self.tag_client.get_all_tags() +``` + +#### GET - GET tag by id + +```python +def test_get_tag_by_id(self): + get_tag = self.tag_client.get_tag_by_id(tags_id = self.tag_obj.unique_id) +``` + +#### PATCH - Update tag by id + +```python +def test_update_tag(self): + update_tag = self.tag_client.update_tag(tag=self.tag_obj, name="updated name", color="green") +``` + +#### DEL - Delete tag + +```python +def test_delete_tag(self): + delete_tag = self.tag_client.delete_tag(tag=self.tag_obj) +``` + +## Task templates + +```python +class SDKTemplateClient(SDKTeamsClient): + def __init__(self): + super().__init__() + self.teams_client = TeamsClient(client=self.client) + self.account_member_client = AccountMemberClient(client=self.client) + self.team_obj = self.teams_client.create_team(name=f"Integration Team - {self.datetime_timestamp}") + + self.tag_client = self.teams_client.get_tag_client(self.team_obj) + + self.task_template_client = self.teams_client.get_task_template_client(self.team_obj) +``` + +#### POST - Create a task template + +```python +def test_create_template(self): + self.task_template_obj = self.task_template_client.create_task_template( + name="Test Task Template", summary="Test Task Template" + ) +``` + +#### GET - Get all task templates + +```python +def test_get_all_task_templates(self): + get_all_task_templates = self.task_template_client.get_all_task_template() +``` + +#### GET - Get task templates by id + +```python +def test_get_task_template_by_id(self): + get_task_template_by_id = self.task_template_client.get_task_template_by_id( + task_template_id=self.task_template_obj.unique_id + ) +``` + +#### PATCH - Update the task template + +```python +def test_update_task_template(self): + update_task_template = self.task_template_client.update_task_template( + task_template = self.task_template_obj, name="Test Task Template Updated" + ) +``` + +#### DEL - Delete the task template + +```python +def test_delete_task_template(self): + delete_task_template = self.task_template_client.delete_task_template(task_template = self.task_template_obj) ``` -Refer the comments under each function for a detailed description of it's parameters. -It is important to note that each function returns a urllib3.response.HTTPResponse object. -## Running tests +# Running tests -There is a sample skeleton code in bin/. Add your access token to it and modify the object and function name for testing purposes. +There is a sample skeleton code in tests/. Add your access token to it and modify the object and function name for testing purposes. -## License +# License This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details diff --git a/setup.py b/setup.py index ceab627..c8b3847 100644 --- a/setup.py +++ b/setup.py @@ -2,12 +2,57 @@ setup( name="zenduty-api", - version="0.8", + version="1.0.0", description="Python SDK wrapper for the Zenduty API", - long_description="Python SDK wrapper for the Zenduty API", + long_description=""" + # Zenduty Python SDK + + The **Zenduty Python SDK** provides a seamless way to integrate with Zenduty's powerful incident management platform. Whether you're building automated workflows, custom dashboards, or advanced monitoring solutions, this SDK offers the tools you need to connect with Zenduty's API endpoints efficiently. + + ## About Zenduty + + Zenduty is a cutting-edge incident management platform designed to help teams resolve incidents faster, smarter, and with greater ease. By leveraging advanced automation, Zenduty ensures that developers and engineers stay in control during high-pressure scenarios. Zenduty empowers teams to: + + - Proactively identify and resolve issues before they escalate. + - Collaborate effectively to minimize downtime and service disruptions. + - Improve reliability and customer satisfaction with actionable insights. + + ## Why Use the Zenduty Python SDK? + + The Zenduty Python SDK is crafted for developers looking to integrate their applications or services with Zenduty. It simplifies the process of interacting with Zenduty's API and supports robust and scalable integrations for a variety of use cases, including: + + - Triggering, acknowledging, and resolving incidents programmatically. + - Accessing detailed analytics and reporting data. + - Managing services, teams, schedules, and escalation policies. + - Building automation pipelines for incident notifications and workflows. + + ## Key Features + + - **Ease of Use**: Intuitive methods and structures to reduce development time. + - **Comprehensive API Support**: Full access to Zenduty's endpoints for managing incidents, teams, schedules, and more. + - **Scalability**: Designed to handle complex workflows and high-volume environments. + - **Compatibility**: Works seamlessly with modern Python environments and frameworks. + + ## Installation + + You can install the Zenduty Python SDK quickly via pip: + + ```sh + $ pip install zenduty-api + ``` + + or you may grab the latest source code from GitHub: + + ```sh + $ git clone https://github.com/Zenduty/zenduty-python-sdk + ``` + + ## Docs + Please refer this link to understand the SDK better. https://github.com/Zenduty/zenduty-python-sdk + """, long_description_content_type="text/x-rst", - author="Vishwa Krishnakumar", - author_email="vishwa@yellowant.com", + author="Javeed Yara", + author_email="javeed@zenduty.com", packages=find_packages(), install_requires=[ "requests==2.32.3", @@ -15,7 +60,18 @@ "six==1.9.0", "charset-normalizer==3.3.2", "idna==3.7", - "certifi==2024.7.4" + "certifi==2024.7.4", + "regex==2024.11.6", + ], + url="https://github.com/Zenduty/zenduty-python-sdk", + license="MIT", + classifiers=[ + "Development Status :: 4 - Beta", # Update based on your package's status + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.9", # Specify Python versions supported ], - scripts=["bin/client.py"], -) \ No newline at end of file + python_requires=">=3.9", # Specify the minimum Python version + scripts=["bin/client.py"], # Include any scripts you want to make executable +) diff --git a/tests/test_sdk_automated.py b/tests/test_sdk_automated.py new file mode 100644 index 0000000..42387fc --- /dev/null +++ b/tests/test_sdk_automated.py @@ -0,0 +1,926 @@ +import sys +import uuid +import time +import pytest +import random +import string +import zenduty +from pathlib import Path +from datetime import datetime, timedelta + +# Importing V2 Clients +from zenduty.apiV2.client import ZendutyClient + +# Importing V2 Internal Clients +from zenduty.apiV2.teams import TeamsClient +from zenduty.apiV2.events import EventClient +from zenduty.apiV2.incidents import IncidentClient +from zenduty.apiV2.events.router import RouterClient +from zenduty.apiV2.incidents.notes import IncidentNoteClient +from zenduty.apiV2.incidents.tags import IncidentTagClient +from zenduty.apiV2.accounts.roles import AccountRoleClient +from zenduty.apiV2.accounts.members import AccountMemberClient +from zenduty.apiV2.teams.escalation_policies.rules import RuleBuilder +from zenduty.apiV2.teams.escalation_policies.targets import TargetBuilder +from zenduty.apiV2.authentication.zenduty_credential import ZendutyCredential + + +class SDKTestingClient: + @classmethod + def setup_class(cls): + cls.cred = ZendutyCredential("e3464dbec1590e0c226685e156f40ed541c3b715") + cls.client = ZendutyClient( + credential=cls.cred, use_https=True + ) # defaults to default service endpoint zenduty.com + cls.datetime_timestamp = cls.datetime_timestamp() + + @staticmethod + def datetime_timestamp(): + return datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + @staticmethod + def generate_uuid() -> str: + # generate a random UUID + return str(uuid.uuid4()) + + +@pytest.mark.teams +class TestSDKTeamsClient(SDKTestingClient): + @classmethod + def setup_class(cls): + super().setup_class() + cls.team_ids = [] + cls.team_members = [] + cls.team_member_unique_id = [] + cls.teams_client = TeamsClient(client=cls.client) + cls.team_member_id = "773c69f5-78f2-42ca-b3d9-b" + cls.invite_url = "https://zenduty.com/api/invite/accept/" + cls.test_team_name = f"Team - {cls.datetime_timestamp}" + + @staticmethod + def create_team(self): + create_team = self.teams_client.create_team(self.test_team_name) + return create_team + + @staticmethod + def delete_team(self): + delete_team = self.teams_client.delete_team(self.team_ids[0]) + + # Teams testing + def test_create_team(self): + # Team1 is the name of the team and that is the payload + create_team = self.teams_client.create_team(self.test_team_name) + self.team_ids.append(create_team) + assert create_team.name == self.test_team_name + time.sleep(2) + + def test_find_team_by_id(self): + find_team = self.teams_client.find_team_by_id(self.team_ids[0].unique_id) + assert find_team.unique_id == self.team_ids[0].unique_id + time.sleep(2) + + def test_update_teams(self): + update_teams = self.teams_client.update_team( + self.team_ids[0], name=f"Updated Team Name - {self.datetime_timestamp}" + ) + assert ( + str(update_teams.name) == f"Updated Team Name - {self.datetime_timestamp}" + ) + + def test_list_team_member(self): + list_team_members = self.teams_client.list_team_members(self.team_ids[0]) + assert list_team_members[0].team == self.team_ids[0].unique_id + time.sleep(2) + + def test_add_team_member(self): + add_team_member = self.teams_client.add_team_member( + self.team_ids[0], username=self.team_member_id + ) + assert str(add_team_member.team) == str(self.team_ids[0].unique_id) + time.sleep(2) + + def test_find_team_member(self): + list_team_members = self.teams_client.list_team_members(self.team_ids[0]) + self.team_member_unique_id.append(list_team_members[1].unique_id) + find_team_member = self.teams_client.find_team_member( + self.team_ids[0], member_unique_id=self.team_member_unique_id[0] + ) + assert str(find_team_member.unique_id) == str(self.team_member_unique_id[0]) + time.sleep(2) + + def test_update_team_member(self): + update_team_member = self.teams_client.update_team_member( + self.team_ids[0], member_id=self.team_member_unique_id[0], role=1 + ) + assert update_team_member.role == 1 and str( + update_team_member.unique_id + ) == str(self.team_member_unique_id[0]) + time.sleep(2) + + def test_delete_team_member(self): + delete_team_member = self.teams_client.delete_team_member( + self.team_ids[0], member_id=self.team_member_unique_id[0] + ) + time.sleep(2) + + def test_fetch_team_permissions(self): + team_permissions = self.teams_client.fetch_team_permissions(self.team_ids[0]) + + def update_team_permissions(self): + updated_team_permissions = self.teams_client.update_team_permissions( + permissions=["service_read"], team=self.team_ids[0] + ) + assert "service_read" in updated_team_permissions + + def test_delete_teams(self): + delete_teams = self.teams_client.delete_team(self.team_ids[0]) + + +@pytest.mark.accountmembers +class TestSDKAccountMembersClient(TestSDKTeamsClient): + @classmethod + def setup_class(cls): + super().setup_class() + cls.account_member_obj = [] + cls.account_member_team_ids = [] + cls.account_member_ids = [] + cls.teams_client = TeamsClient(client=cls.client) + cls.account_member_client = AccountMemberClient(client=cls.client) + + def test_account_members_invite(self): + create_team = self.teams_client.create_team( + name="Random Testing Team" + self.datetime_timestamp + ) + self.account_member_team_ids.append(create_team.unique_id) + + test_email = f"john.doe.{random.randint(2,10000000000000000000000)}@zenduty.com" + + account_member_invite = self.account_member_client.invite( + team_id=self.account_member_team_ids[0], + first_name="John", + last_name="doe", + role=3, + email=test_email, + ) + self.account_member_obj.append(account_member_invite) + self.account_member_ids.append(account_member_invite.user.username) + assert account_member_invite.user.email == test_email + + def test_account_member_update(self): + test_first_name = f"Jane {random.randint(2,10000000000000000000000)}" + # updated the email + update_account_member = self.account_member_client.update_account_member( + account_member=self.account_member_obj[0], + first_name=test_first_name, + last_name=f"Doe {random.randint(2,10000000000000000000000)}", + role=2, + ) + + assert update_account_member.user.first_name == test_first_name + + def test_get_account_member(self): + account_member = self.account_member_client.get_account_member( + account_member_id=self.account_member_ids[0] + ) + assert account_member.user.username == self.account_member_ids[0] + + def test_get_all_account_members(self): + account_members = self.account_member_client.get_all_members() + + def test_delete_account_member(self): + delete_account_member = self.account_member_client.delete_account_member( + account_member=self.account_member_obj[0] + ) + + +@pytest.mark.accountroles +class TestSDKAccountRolesClient(SDKTestingClient): + @classmethod + def setup_class(cls): + super().setup_class() + cls.account_role_obj = [] + cls.account_role_ids = [] + cls.account_role_client = AccountRoleClient(client=cls.client) + + def test_create_account_role(self): + test_name = f"Account Role - {self.datetime_timestamp}" + create_account_role = self.account_role_client.create_account_role( + name=test_name, + description="Account Role Description", + permissions=["sla_read"], + ) + + self.account_role_obj.append(create_account_role) + self.account_role_ids.append(create_account_role.unique_id) + + assert create_account_role.name == test_name + + time.sleep(2) + + def test_get_account_role(self): + get_account_role = self.account_role_client.get_account_role( + account_role_id=self.account_role_ids[0] + ) + + assert str(get_account_role.unique_id) == str(self.account_role_ids[0]) + time.sleep(2) + + def test_list_account_roles(self): + list_account_roles = self.account_role_client.list_account_roles() + time.sleep(2) + + def test_update_account_role(self): + test_name = f"Updated Account Role - {self.datetime_timestamp}" + update_account_role = self.account_role_client.update_account_role( + account_role=self.account_role_obj[0], + name=test_name, + description="Updated Account Role Description", + permissions=["sla_read"], + ) + + assert update_account_role.name == test_name + time.sleep(2) + + def test_delete_account_role(self): + delete_account_role = self.account_role_client.delete_account_role( + account_role=self.account_role_obj[0] + ) + time.sleep(2) + + +@pytest.mark.GER +class TestSDKGERClients(SDKTestingClient): + @classmethod + def setup_class(cls): + super().setup_class() + cls.router_ids = [] + cls.router_obj = [] + cls.router_client = RouterClient(client=cls.client) + cls.router_name = f"Router - {cls.datetime_timestamp}" + + def test_create_router(self): + create_router = self.router_client.create_router( + name=self.router_name, + description="Router Description", + ) + self.router_ids.append(create_router.unique_id) + self.router_obj.append(create_router) + assert str(create_router.name) == f"Router - {self.datetime_timestamp}" + time.sleep(2) + + def test_list_routers(self): + list_router = self.router_client.get_all_routers() + for router in list_router: + if str(router.name) == f"Router - {self.datetime_timestamp}": + assert str(router.name) == f"Router - {self.datetime_timestamp}" + break + time.sleep(2) + + def get_all_routers(self): + return self.router_client.get_all_routers() + + def test_get_router_by_id(self): + find_router = self.router_client.get_router_by_id(self.router_obj[0].unique_id) + assert str(find_router.unique_id) == str(self.router_ids[0]) + time.sleep(2) + + def test_update_router(self): + update_router = self.router_client.update_router( + router=self.router_obj[0], + name="Updated Router Name", + description="Updated Router Description", + ) + assert str(update_router.name) == "Updated Router Name" + + time.sleep(2) + + def test_delete_router(self): + delete_router = self.router_client.delete_router(self.router_obj[0]) + time.sleep(2) + + +@pytest.mark.events +class TestSDKEventsClient(SDKTestingClient): + @classmethod + def setup_class(cls): + super().setup_class() + cls.event_ids = [] + cls.event_client = EventClient(client=cls.client) + cls.event_name = f"Event - {cls.datetime_timestamp}" + + def test_get_router_client(self): + get_router = self.event_client.get_router_client() + + def test_create_event(self): + create_event = self.event_client.create_event( + integration_key="f86e6ade-f987-4cfc-b047-9ce9ca794b41", + alert_type="info", + message="This is info alert", + summary="This is the incident summary111", + entity_id=123455, + payload={ + "status": "ACME Payments are failing", + "severity": "1", + "project": "kubeprod", + }, + urls=[ + { + "link_url": "https://www.example.com/alerts/12345/", + "link_text": "Alert URL", + } + ], + ) + + assert ( + create_event.integration_object.integration_key + == "f86e6ade-f987-4cfc-b047-9ce9ca794b41" + ) + + +@pytest.mark.escalationpolicy +class TestSDKEscalationPolicyClient(TestSDKTeamsClient): + @classmethod + def setup_class(cls): + super().setup_class() + cls.escalation_policy_ids = [] + cls.account_member_ids = [] + cls.team_ids = [] + cls.create_escalation_policy_obj = [] + cls.uuid = cls.generate_uuid() + cls.teams_client = TeamsClient(client=cls.client) + cls.account_member_client = AccountMemberClient(client=cls.client) + cls.team_ids.append(cls.create_team(cls)) + # create teams fails because of this. + cls.team_by_id = cls.teams_client.find_team_by_id( + team_id="999a17ed-c7c3-4860-9024-d11c18fa5fa4" + ) + cls.escalation_policy_client = cls.teams_client.get_escalation_policy_client( + cls.team_by_id + ) + cls.ep_name = f"EP - {cls.datetime_timestamp}" + + @staticmethod + def generate_uuid() -> str: + return str(uuid.uuid4()) + + def test_create_escalation_policy(self): + + self.rule_build = [ + { + "delay": 0, + "targets": [ + {"target_type": 2, "target_id": "3544118d-fbf5-41e5-ae6c-5"} + ], + "position": 1, + } + ] + create_escalation_policy = self.escalation_policy_client.create_esp( + self.ep_name, rules=self.rule_build + ) + + # Appending the unique_id to the escalation_policy_ids list + self.escalation_policy_ids.append(create_escalation_policy.unique_id) + self.create_escalation_policy_obj.append(create_escalation_policy) + assert create_escalation_policy.name == self.ep_name + time.sleep(2) + + def test_get_esp_by_id(self): + self.get_esp_by_id = self.escalation_policy_client.get_esp_by_id( + esp_id=self.escalation_policy_ids[0] + ) + assert self.get_esp_by_id.name == self.ep_name + time.sleep(2) + + def test_update_esp(self): + self.rule_build = [ + { + "delay": 0, + "targets": [ + {"target_type": 2, "target_id": "3544118d-fbf5-41e5-ae6c-5"} + ], + "position": 1, + } + ] + update_esp = self.escalation_policy_client.update_esp( + esp=self.create_escalation_policy_obj[0], + name=f"Test Updated - {self.datetime_timestamp}", + rules=self.rule_build, + ) + + assert update_esp.name == f"Test Updated - {self.datetime_timestamp}" + time.sleep(2) + + def test_get_all_policies(self): + all_esp = self.escalation_policy_client.get_all_policies() + time.sleep(2) + + def test_delete_esp(self): + delete_esp = self.escalation_policy_client.delete_esp( + esp=self.create_escalation_policy_obj[0] + ) + time.sleep(2) + + +@pytest.mark.maintenance +class TestSDKMaintenanceClient(TestSDKTeamsClient): + @classmethod + def setup_class(cls): + super().setup_class() + cls.maintenance_ids = [] + cls.account_member_ids = [] + cls.team_ids = [] + cls.maintenance_obj = [] + cls.uuid = cls.generate_uuid() + cls.teams_client = TeamsClient(client=cls.client) + cls.team_ids.append(cls.create_team(cls)) + cls.team_by_id = cls.teams_client.find_team_by_id( + team_id="999a17ed-c7c3-4860-9024-d11c18fa5fa4" + ) + cls.maintenance_client = cls.teams_client.get_maintenance_client(cls.team_by_id) + cls.maintenance_name = f"Maintenance Mode - {cls.datetime_timestamp}" + + def test_create_maintenance(self): + create_maintenance = self.maintenance_client.create_team_maintenance( + name=self.maintenance_name, + start_time="2026-07-08T18:06:00", + end_time="2026-07-08T18:06:00", + service_ids=["a91a3a00-8de9-472c-ad2e-61e7c89db062"], + ) + self.maintenance_obj.append(create_maintenance) + self.maintenance_ids.append(create_maintenance.unique_id) + assert create_maintenance.name == self.maintenance_name + time.sleep(2) + + def test_get_all_maintenance(self): + get_all_maintenance = self.maintenance_client.get_all_maintenance() + + time.sleep(2) + + def test_get_maintenance_by_id(self): + get_maintenance_by_id = self.maintenance_client.get_maintenance_by_id( + maintenance_id=self.maintenance_ids[0] + ) + + time.sleep(2) + + def test_update_maintenance(self): + update_maintenance = self.maintenance_client.update_maintenance( + maintenance=self.maintenance_obj[0], + name="Updated Maintenance Name", + start_time="2026-07-08T18:06:00", + end_time="2026-07-08T18:06:00", + service_ids=["a91a3a00-8de9-472c-ad2e-61e7c89db062"], + ) + + time.sleep(2) + + def test_delete_maintenance(self): + delete_maintenance = self.maintenance_client.delete_maintenance( + maintenance=self.maintenance_obj[0] + ) + + time.sleep(2) + + +@pytest.mark.incidents +class TestSDKIncidentsClient(SDKTestingClient): + @classmethod + def setup_class(cls): + super().setup_class() + cls.incident_ids = [] + cls.incident_number = [] + cls.incident_notes_list = [] + cls.incident_tags_list = [] + cls.incident_client = IncidentClient(client=cls.client) + cls.incident_name = f"Incident - {cls.datetime_timestamp}" + cls.incident_notes = f"Incident Notes - {cls.datetime_timestamp}" + cls.incident_tags = f"Incident Tags - {cls.datetime_timestamp}" + + def test_create_incident(self): + create_incident = self.incident_client.create_incident( + title=self.incident_name, service="a91a3a00-8de9-472c-ad2e-61e7c89db062" + ) + + assert ( + str(create_incident.title) == self.incident_name + and str(create_incident.service) == "a91a3a00-8de9-472c-ad2e-61e7c89db062" + ) + + self.incident_ids.append(create_incident.unique_id) + self.incident_number.append(create_incident.incident_number) + time.sleep(2) + + # check here for incident notes and tags + def test_create_incident_note(self): + # Creating a Incident Note client + self.note_client = self.incident_client.get_note_client( + incident=self.incident_obj + ) + + # Creating an incident note, attaching it to an incident + create_incident_note = self.note_client.create_incident_note( + note=self.incident_notes + ) + self.incident_notes_list.append(create_incident_note.unique_id) + + assert str(create_incident_note.note) == self.incident_notes + + time.sleep(2) + + def test_get_all_incident_notes(self): + get_all_incident_notes = self.note_client.get_all_incident_notes() + time.sleep(2) + + def test_get_incident_note_by_id(self): + get_incident_note_by_id = self.note_client.get_incident_note_by_id( + incident_note_unique_id=self.incident_note_obj.unique_id + ) + time.sleep(2) + + # get this checked tomorrow + def test_update_incident_note(self): + update_incident_note = self.note_client.update_incident_note( + incident_note=self.incident_note_obj, + note="Updated Incident Note", + ) + + def test_delete_incident_note(self): + delete_incident_note = self.note_client.delete_incident_note( + incident_note=self.incident_note_obj + ) + + # get this checked tomorrow + def test_create_incident_tag(self): + self.tag_client = self.incident_client.get_tags_client(self.incident_number[0]) + + create_incident_tag = self.tag_client.create_tag( + team_tag=self.incident_number[0] + ) + + def test_get_all_tags(self): + get_all_tags = self.tag_client.get_all_tags() + + def test_get_tag_by_id(self): + get_tag_by_id = self.tag_client.get_tag_by_id(self.incident_tags[0]) + + def test_delete_incident_tag(self): + delete_incident_tag = self.tag_client.delete_tag(self.incident_tags[0]) + + def test_get_all_incidents(self): + get_all_incidents = self.incident_client.get_all_incidents(page=1) + + for incident in get_all_incidents: + if str(incident["title"]) == self.incident_name: + assert str(incident["title"]) == self.incident_name + break + time.sleep(2) + + def test_get_alerts_by_incident(self): + get_alerts_by_incident = self.incident_client.get_alerts_for_incident( + incident_number=self.incident_obj.incident_number + ) + time.sleep(2) + + def test_update_incident(self): + update_incident = self.incident_client.update_incident( + incident_id=self.incident_obj.unique_id, + title="Updated Incident Name", + status=3, + service="a91a3a00-8de9-472c-ad2e-61e7c89db062", + ) + + assert ( + int(update_incident.status) == 3 + and str(update_incident.title) == "Updated Incident Name" + and str(update_incident.unique_id) == str(self.incident_ids[0]) + and str(update_incident.incident_number) == str(self.incident_number[0]) + ) + + +@pytest.mark.postmortem +class TestSDKPostMortemClient(TestSDKTeamsClient): + @classmethod + def setup_class(cls): + super().setup_class() + cls.team_ids = [] + cls.incident_ids = [] + cls.postmortem_ids = [] + cls.account_member_ids = [] + cls.incident_name = "blahblah" + cls.uuid = cls.generate_uuid() + cls.teams_client = TeamsClient(client=cls.client) + cls.team_ids.append(cls.create_team(cls)) + cls.team_by_id = cls.teams_client.find_team_by_id( + team_id="999a17ed-c7c3-4860-9024-d11c18fa5fa4" + ) + cls.incident_client = IncidentClient(client=cls.client) + cls.incident_name = f"Incident - {cls.datetime_timestamp}" + cls.postmortem_client = cls.teams_client.get_postmortem_client(cls.team_by_id) + cls.postmortem_name = f"Postmortem - {cls.datetime_timestamp}" + + def test_create_postmortem(self): + # Create the Incident + create_incident = self.incident_client.create_incident( + title=self.incident_name, service="a91a3a00-8de9-472c-ad2e-61e7c89db062" + ) + + # Create the Postmortem + create_postmortem = self.postmortem_client.create_postmortem( + author="3544118d-fbf5-41e5-ae6c-5", + incidents=[create_incident.unique_id], + title="Test Postmortem", + ) + + # Appending the unique id to the postmortem list. + self.postmortem_ids.append(create_postmortem.unique_id) + self.incident_ids.append(create_incident.unique_id) + + def test_get_postmortem_by_id(self): + self.postmortem_by_id = self.postmortem_client.get_postmortem_by_id( + postmortem_id=self.postmortem_ids[0] + ) + + assert self.postmortem_by_id.title == "Test Postmortem" + + def test_update_postmortem(self): + update_postmortem = self.postmortem_client.update_postmortem( + self.postmortem_by_id, + author="3544118d-fbf5-41e5-ae6c-5", + incidents=[self.incident_ids[0]], + title="Test Postmortem Updated", + ) + assert update_postmortem.title == "Test Postmortem Updated" + + def test_delete_postmortem(self): + delete_postmortem = self.postmortem_client.delete_postmortem( + self.postmortem_by_id + ) + + # Resolve the incident + resolve_incident = self.incident_client.update_incident( + self.postmortem_obj, + title=self.incident_name, + status=3, + ) + + +@pytest.mark.priorities +class TestSDKPrioritiesClient(TestSDKTeamsClient): + @classmethod + def setup_class(cls): + super().setup_class() + cls.team_ids = [] + cls.priority_ids = [] + cls.account_member_ids = [] + cls.uuid = cls.generate_uuid() + cls.teams_client = TeamsClient(client=cls.client) + cls.team_ids.append(cls.create_team(cls)) + cls.team_by_id = cls.teams_client.find_team_by_id( + team_id="999a17ed-c7c3-4860-9024-d11c18fa5fa4" + ) + cls.priority_client = cls.teams_client.get_priority_client(cls.team_by_id) + cls.priority_name = f"Priority - {cls.datetime_timestamp}" + + def test_create_priority(self): + create_priority = self.priority_client.create_priority( + name=self.priority_name, + description="Priority Description", + color="red", + ) + + self.priority_ids.append(create_priority.unique_id) + assert create_priority.name == self.priority_name + time.sleep(2) + + def test_get_all_priorities(self): + get_all_priorities = self.priority_client.get_all_priorities() + time.sleep(2) + + def test_get_priority_by_id(self): + self.priority_by_id = self.priority_client.get_priority_by_id( + priority_id=self.priority_ids[0] + ) + + assert self.priority_by_id.name == self.priority_name + + def test_update_priority(self): + update_priority = self.priority_client.update_priority( + self.priority_obj, + name="Test Priority Updated", + description="Test Priority", + ) + + assert update_priority.name == "Test Priority Updated" + time.sleep(2) + + def test_delete_priority(self): + delete_priority = self.priority_client.delete_priority(self.priority_obj) + time.sleep(2) + + +@pytest.mark.roles +class TestSDKRolesClient(TestSDKTeamsClient): + @classmethod + def setup_class(cls): + super().setup_class() + cls.team_ids = [] + cls.role_ids = [] + cls.account_member_ids = [] + cls.uuid = cls.generate_uuid() + cls.teams_client = TeamsClient(client=cls.client) + cls.team_ids.append(cls.create_team(cls)) + cls.team_by_id = cls.teams_client.find_team_by_id( + team_id="999a17ed-c7c3-4860-9024-d11c18fa5fa4" + ) + cls.role_client = cls.teams_client.get_incident_role_client(cls.team_by_id) + cls.role_name = f"Role - {cls.datetime_timestamp}" + + def test_create_role(self): + self.create_role = self.role_client.create_incident_role( + title="Test Role", + description="Test Role", + rank=1, + ) + + self.role_ids.append(self.create_role.unique_id) + assert self.create_role.title == "Test Role" + time.sleep(2) + + def test_get_incident_role_by_id(self): + self.get_role_by_id = self.role_client.get_incident_role_by_id( + role_id=self.role_ids[0] + ) + assert self.get_role_by_id.title == "Test Role" + time.sleep(2) + + def test_update_incident_role(self): + self.update_role = self.role_client.update_incident_role( + role=self.role_obj, + title="Test Role Updated", + ) + assert self.update_role.title == "Test Role Updated" + time.sleep(2) + + def test_delete_incident_role(self): + self.delete_role = self.role_client.delete_incident_role( + role=self.role_obj, + ) + time.sleep(2) + + +@pytest.mark.schedules +class TestSDKSchedulesClient(TestSDKTeamsClient): + @classmethod + def setup_class(cls): + super().setup_class() + cls.team_ids = [] + cls.schedules_ids = [] + cls.account_member_ids = [] + cls.schedules_obj = [] + cls.uuid = cls.generate_uuid() + cls.teams_client = TeamsClient(client=cls.client) + cls.team_ids.append(cls.create_team(cls)) + cls.team_by_id = cls.teams_client.find_team_by_id( + team_id="999a17ed-c7c3-4860-9024-d11c18fa5fa4" + ) + cls.schedules_client = cls.teams_client.get_schedule_client(cls.team_by_id) + cls.schedules_name = f"Schedules - {cls.datetime_timestamp}" + cls.layers = [ + { + "name": "Layer 1", + "is_active": True, + "restriction_type": 0, + "restrictions": [], + "rotation_start_time": "2025-07-29T03:30:00.000Z", + "rotation_end_time": None, + "shift_length": 86400, + "users": [ + { + "user": "3544118d-fbf5-41e5-ae6c-5", + "position": 1, + } + ], + } + ] + + cls.overrides = [ + { + "name": "", + "user": "3544118d-fbf5-41e5-ae6c-5", + "start_time": "2024-07-29T11:54:34.745000Z", + "end_time": "2024-07-29T18:29:59.999000Z", + } + ] + + def test_create_schedule(self): + create_schedule = self.schedules_client.create_schedule( + name=self.schedules_name, + timezone="Asia/Kolkata", + layers=self.layers, + overrides=self.overrides, + ) + + self.schedules_ids.append(create_schedule.unique_id) + self.schedules_obj.append(create_schedule) + assert create_schedule.name == self.schedules_name + time.sleep(2) + + def test_get_all_schedules(self): + get_all_schedules = self.schedules_client.get_all_schedules() + time.sleep(2) + + def test_get_schedule_by_id(self): + self.get_schedule_by_id = self.schedules_client.get_schedule_by_id( + schedule_id=self.schedules_ids[0] + ) + + assert self.get_schedule_by_id.name == self.schedules_name + time.sleep(2) + + def test_update_schedule(self): + update_schedule = self.schedules_client.update_schedule( + schedule=self.schedules_obj[0], + name="Test Schedule Updated", + ) + assert update_schedule.name == "Test Schedule Updated" + + def test_delete_schedule(self): + delete_schedule = self.schedules_client.delete_schedule( + schedule=self.schedules_obj[0] + ) + + +@pytest.mark.services +class TestSDKServicesClient(TestSDKTeamsClient): + @classmethod + def setup_class(cls): + super().setup_class() + cls.team_ids = [] + cls.sla_ids = [] + cls.priority_ids = [] + cls.escalation_policy_ids = [] + cls.service_ids = [] + # Making the Teams Client + cls.teams_client = TeamsClient(client=cls.client) + cls.team_ids.append(cls.create_team(cls)) + # Making the Service Client + cls.service_client = cls.teams_client.get_service_client(cls.team_ids[0]) + cls.team_by_id = cls.teams_client.find_team_by_id( + team_id="999a17ed-c7c3-4860-9024-d11c18fa5fa4" + ) + cls.escalation_policy_client = cls.teams_client.get_escalation_policy_client( + cls.team_by_id + ) + cls.priority_client = cls.teams_client.get_priority_client(cls.team_by_id) + cls.sla_client = cls.teams_client.get_sla_client(cls.team_by_id) + # Making the names + cls.ep_name = f"EP - {cls.datetime_timestamp}" + cls.priority_name = f"Priority - {cls.datetime_timestamp}" + cls.sla_name = f"SLA - {cls.datetime_timestamp}" + cls.service_name = f"Service - {cls.datetime_timestamp}" + + def test_create_service(self): + # Create the escalation policy + self.rule_build = [ + { + "delay": 0, + "targets": [ + {"target_type": 2, "target_id": "3544118d-fbf5-41e5-ae6c-5"} + ], + "position": 1, + } + ] + create_escalation_policy = self.escalation_policy_client.create_esp( + self.ep_name, rules=self.rule_build + ) + + # Appending the unique_id to the escalation_policy_ids list + self.escalation_policy_ids.append(str(create_escalation_policy.unique_id)) + print("ids", self.escalation_policy_ids) + time.sleep(2) + # Create the Priority + create_priority = self.priority_client.create_priority( + name=self.priority_name, + description="Priority Description", + color="red", + ) + + # Appending the unique_id to the priority_ids list + self.priority_ids.append(create_priority.unique_id) + time.sleep(2) + + # Create the SLA + create_sla = self.sla_client.create_sla(name="Test SLA", escalations=[]) + + # Appending the unique_id to the sla_ids list + self.sla_ids.append(create_sla.unique_id) + + time.sleep(2) + + # Finally create the service + create_service = self.service_client.create_service( + name=f"Test Service - {self.datetime_timestamp}", + escalation_policy=self.escalation_policy_ids[0], + team_priority=str(self.priority_ids[0]), + sla=str(self.sla_ids[0]), + ) diff --git a/zenduty/apiV2/teams/maintenance/models.py b/zenduty/apiV2/teams/maintenance/models.py index 4643973..6b132de 100644 --- a/zenduty/apiV2/teams/maintenance/models.py +++ b/zenduty/apiV2/teams/maintenance/models.py @@ -23,6 +23,7 @@ class TeamMaintenance(JsonSerializable): name: str time_zone: str repeat_until: Optional[datetime] + maintenance_template: str def __init__( self, @@ -35,6 +36,7 @@ def __init__( name: str, time_zone: str, repeat_until: Optional[int], + maintenance_template: str, ) -> None: self.unique_id = unique_id if type(unique_id) is not str else UUID(unique_id) self.start_time = ( @@ -66,3 +68,4 @@ def __init__( if type(repeat_until) is datetime or repeat_until is None else datetime.fromisoformat(repeat_until.replace("Z", "+00:00")) ) + self.maintenance_template = maintenance_template diff --git a/zenduty/apiV2/teams/oncall/__init__.py b/zenduty/apiV2/teams/oncall/__init__.py new file mode 100644 index 0000000..e69de29