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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 27 additions & 17 deletions did/plugins/zammad.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,10 @@
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime
from datetime import date, datetime
from typing import Any, Optional, cast

from did.base import Config, ReportError, get_token
from did.base import Config, ReportError, User, get_token
from did.stats import Stats, StatsGroup
from did.utils import listed, log, pretty

Expand All @@ -35,7 +36,7 @@ class Zammad():
""" Zammad Investigator """
# pylint: disable=too-few-public-methods

def __init__(self, url, token):
def __init__(self, url: str, token: Optional[str]) -> None:
Comment thread
kwk marked this conversation as resolved.
""" Initialize url and headers """
self.url = url.rstrip("/")
if token is not None:
Expand All @@ -45,22 +46,22 @@ def __init__(self, url, token):

self.token = token

def perform_search(self, query: str) -> dict:
def perform_search(self, query: str) -> dict[str, Any]:
""" Perform Zammad query """
url = f"{self.url}/{query}"
log.debug("Zammad query: %s", url)
try:
request = urllib.request.Request(url, headers=self.headers)
with urllib.request.urlopen(request) as response:
log.debug("Response headers:\n%s", str(response.info()).strip())
return json.loads(response.read())
return cast(dict[str, Any], json.loads(response.read()))
except urllib.error.URLError as error:
log.debug(error)
raise ReportError(
f"Zammad search on {self.url} failed.") from error

def search(self, query: str) -> dict:
result = self.perform_search(query)["assets"]
def search(self, query: str) -> dict[str, Any]:
result: dict[str, Any] = self.perform_search(query)["assets"]
try:
result = result["Ticket"]
except KeyError:
Expand All @@ -69,8 +70,9 @@ def search(self, query: str) -> dict:
log.data(pretty(result))
return result

def get_articles(self, ticket_id):
result = self.perform_search("/ticket_articles/by_ticket/" + str(ticket_id))
def get_articles(self, ticket_id: str) -> list[dict[str, Any]]:
result: list[dict[str, Any]] = self.perform_search(
Comment thread
kwk marked this conversation as resolved.
f"/ticket_articles/by_ticket/{ticket_id}")["assets"]
log.debug("Result: %s fetched", listed(len(result), "item"))
log.data(pretty(result))
return result
Expand All @@ -80,16 +82,16 @@ def get_articles(self, ticket_id):
# Ticket
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

class Ticket():
class Ticket:
""" Zammad Ticket """
# pylint: disable=too-few-public-methods

def __init__(self, data):
def __init__(self, data: dict[str, Any]) -> None:
self.data = data
self.title = data["title"]
self.id = data["id"]

def __str__(self):
def __str__(self) -> str:
""" String representation """
return f"{str(self.id).zfill(PADDING)} - {self.title}"

Expand All @@ -101,7 +103,11 @@ def __str__(self):
class TicketsUpdated(Stats):
""" Tickets updated """

def fetch(self):
def fetch(self) -> None:
self.parent: ZammadStats
self.user: User
if self.options is None or self.user is None:
raise RuntimeError("ZammadStats not properly initialized")
Comment thread
sandrobonazzola marked this conversation as resolved.
log.info("Searching for tickets updated by %s", self.user)
search = (
f"article.from:\"{self.user.name}\" and "
Expand All @@ -113,9 +119,9 @@ def fetch(self):
until = self.options.until.date
for _, ticket in self.parent.zammad.search(query).items():
for article in self.parent.zammad.get_articles(ticket["id"]):
updated_at = datetime.fromisoformat(
article["updated_at"].replace('Z', '+00:00')).date()
if (article["created_by"] == self.user.email and
date_str = article["updated_at"].replace('Z', '+00:00')
updated_at: date = datetime.fromisoformat(date_str).date()
Comment thread
sandrobonazzola marked this conversation as resolved.
if (str(article["created_by"]) == self.user.email and
since <= updated_at <= until):
self.stats.append(Ticket(ticket))
break
Expand All @@ -131,7 +137,11 @@ class ZammadStats(StatsGroup):
# Default order
order = 680

def __init__(self, option, name=None, parent=None, user=None):
def __init__(self,
option: str,
name: Optional[str] = None,
parent: Optional[StatsGroup] = None,
user: Optional[User] = None) -> None:
Comment thread
kwk marked this conversation as resolved.
StatsGroup.__init__(self, option, name, parent, user)
config = dict(Config().section(option))
# Check server url
Expand Down
4 changes: 2 additions & 2 deletions tests/plugins/test_zammad.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~


def test_missing_url(caplog: LogCaptureFixture):
def test_missing_url(caplog: LogCaptureFixture) -> None:
""" Missing Zammad url results in Error logged """
# using a generic url just to avoid failing on missing url
did.base.Config(BASIC_CONFIG)
Expand All @@ -32,7 +32,7 @@ def test_missing_url(caplog: LogCaptureFixture):
assert "Skipping section zammad due to error: No zammad url set" in caplog.text


def test_wrong_url(caplog: LogCaptureFixture):
def test_wrong_url(caplog: LogCaptureFixture) -> None:
""" Giving a non-Zammad url results in Exception """
did.base.Config(BASIC_CONFIG + "url=https://www.google.com")
with caplog.at_level(logging.ERROR):
Expand Down