Jason Thistlethwaite
103 lines
4.0 KiB
Python
103 lines
4.0 KiB
Python
import unittest
|
|
|
|
from semantic_index.redmine import RedmineApiSource
|
|
|
|
|
|
class RecordingRedmineSource(RedmineApiSource):
|
|
def __init__(self):
|
|
super().__init__(redmine_url="http://redmine.local", api_key="secret", project_identifier="customer-service")
|
|
self.urls = []
|
|
|
|
def _get_json(self, url):
|
|
self.urls.append(url)
|
|
if url.startswith("http://redmine.local/issues.json"):
|
|
return {"issues": [{"id": 39779}]}
|
|
return {"issue": {"id": 39779, "subject": "Goods return"}}
|
|
|
|
|
|
class PagedRedmineSource(RedmineApiSource):
|
|
def __init__(self):
|
|
super().__init__(redmine_url="http://redmine.local", api_key="secret", project_identifier="customer-service")
|
|
self.urls = []
|
|
|
|
def _get_json(self, url):
|
|
self.urls.append(url)
|
|
if url.startswith("http://redmine.local/issues.json"):
|
|
query = url.split("?", 1)[1]
|
|
params = dict(part.split("=", 1) for part in query.split("&"))
|
|
offset = int(params.get("offset", "0"))
|
|
limit = int(params.get("limit", "0"))
|
|
return {"issues": [{"id": issue_id} for issue_id in range(offset + 1, offset + limit + 1)]}
|
|
issue_id = int(url.split("/issues/", 1)[1].split(".", 1)[0])
|
|
return {"issue": {"id": issue_id, "subject": f"Issue {issue_id}"}}
|
|
|
|
|
|
class DuplicatePagedRedmineSource(RedmineApiSource):
|
|
def __init__(self):
|
|
super().__init__(redmine_url="http://redmine.local", api_key="secret", project_identifier="customer-service")
|
|
|
|
def _get_json(self, url):
|
|
if url.startswith("http://redmine.local/issues.json"):
|
|
query = url.split("?", 1)[1]
|
|
params = dict(part.split("=", 1) for part in query.split("&"))
|
|
offset = int(params.get("offset", "0"))
|
|
if offset == 0:
|
|
return {"issues": [{"id": 1}, {"id": 2}]}
|
|
if offset == 2:
|
|
return {"issues": [{"id": 2}, {"id": 3}]}
|
|
return {"issues": []}
|
|
issue_id = int(url.split("/issues/", 1)[1].split(".", 1)[0])
|
|
return {"issue": {"id": issue_id, "subject": f"Issue {issue_id}"}}
|
|
|
|
|
|
class RedmineApiSourceTest(unittest.TestCase):
|
|
def test_recent_issue_summaries_do_not_fetch_issue_details(self):
|
|
source = RecordingRedmineSource()
|
|
|
|
summaries = list(source.recent_issue_summaries(limit=1))
|
|
|
|
self.assertEqual(39779, summaries[0]["id"])
|
|
self.assertEqual(1, len(source.urls))
|
|
self.assertTrue(source.urls[0].startswith("http://redmine.local/issues.json"))
|
|
|
|
def test_issue_detail_fetches_journals_and_helpdesk(self):
|
|
source = RecordingRedmineSource()
|
|
|
|
detail = source.issue_detail(39779)
|
|
|
|
self.assertEqual(39779, detail["id"])
|
|
self.assertIn("include=journals%2Chelpdesk", source.urls[0])
|
|
|
|
def test_recent_helpdesk_issues_requests_helpdesk_include_with_journals(self):
|
|
source = RecordingRedmineSource()
|
|
|
|
issues = list(source.recent_helpdesk_issues(limit=1))
|
|
|
|
self.assertEqual(39779, issues[0]["id"])
|
|
self.assertIn("include=journals%2Chelpdesk", source.urls[1])
|
|
self.assertIn("subproject_id=%21%2A", source.urls[0])
|
|
|
|
def test_recent_helpdesk_issues_paginates_past_redmine_page_limit(self):
|
|
source = PagedRedmineSource()
|
|
|
|
issues = list(source.recent_helpdesk_issues(limit=250))
|
|
|
|
self.assertEqual(250, len(issues))
|
|
list_urls = [url for url in source.urls if url.startswith("http://redmine.local/issues.json")]
|
|
self.assertEqual(3, len(list_urls))
|
|
self.assertIn("limit=100", list_urls[0])
|
|
self.assertIn("offset=0", list_urls[0])
|
|
self.assertIn("offset=100", list_urls[1])
|
|
self.assertIn("offset=200", list_urls[2])
|
|
|
|
def test_recent_helpdesk_issues_skips_duplicate_issue_ids_across_pages(self):
|
|
source = DuplicatePagedRedmineSource()
|
|
|
|
issues = list(source.recent_helpdesk_issues(limit=3))
|
|
|
|
self.assertEqual([1, 2, 3], [issue["id"] for issue in issues])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|