Skip to content

Commit

Permalink
Add Notice and get_notices tests
Browse files Browse the repository at this point in the history
  • Loading branch information
@benhoyt
benhoyt committed Dec 7, 2023
1 parent 9046b5d commit a1eb569
Show file tree
Hide file tree
Showing 3 changed files with 205 additions and 10 deletions.
6 changes: 3 additions & 3 deletions ops/model.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2736,7 +2736,7 @@ def get_notices(
self,
user_ids: Optional[Iterable[int]] = None,
special_user: Optional[pebble.NoticeSpecialUser] = None,
types: Optional[Iterable[str]] = None,
types: Optional[Iterable[pebble.NoticeType]] = None,
keys: Optional[Iterable[str]] = None,
visibilities: Optional[Iterable[pebble.NoticeVisibility]] = None,
after: Optional[datetime.datetime] = None,
Expand Down Expand Up @@ -3546,8 +3546,8 @@ class LazyNotice:
last_repeated: datetime.datetime
occurrences: int
last_data: Dict[str, str]
repeat_after: Optional[datetime.timedelta] = None
expire_after: Optional[datetime.timedelta] = None
repeat_after: Optional[datetime.timedelta]
expire_after: Optional[datetime.timedelta]

def __init__(self, container: Container, id: str, type: str, key: str):
self._container = container
Expand Down
15 changes: 8 additions & 7 deletions ops/pebble.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1351,7 +1351,7 @@ class Notice:
occurrences: int
"""The number of times one of these notices has occurred."""

last_data: Dict[str, str]
last_data: Dict[str, str] = dataclasses.field(default_factory=dict)
"""Additional data captured from the last occurrence of one of these notices."""

repeat_after: Optional[datetime.timedelta] = None
Expand Down Expand Up @@ -2766,7 +2766,7 @@ def get_notices(
self,
user_ids: Optional[Iterable[int]] = None,
special_user: Optional[NoticeSpecialUser] = None,
types: Optional[Iterable[str]] = None,
types: Optional[Iterable[NoticeType]] = None,
keys: Optional[Iterable[str]] = None,
visibilities: Optional[Iterable[NoticeVisibility]] = None,
after: Optional[datetime.datetime] = None,
Expand All @@ -2789,15 +2789,16 @@ def get_notices(
visibilities
after: filter for notices that were last repeated after this time
"""
query = {}
query: Dict[str, Union[str, List[str]]] = {}
if user_ids is not None:
query['user-ids'] = [str(u) for u in user_ids]
if special_user is not None:
if user_ids is not None:
raise TypeError('may not specify both user_ids and special_user')
query['user-ids'] = special_user.value
if 'user-ids' not in query:
query['user-ids'] = []
query_user_ids = typing.cast(List[str], query['user-ids'])
query_user_ids.append(special_user.value)
if types is not None:
query['types'] = list(types)
query['types'] = [t.value for t in types]
if keys is not None:
query['keys'] = list(keys)
if visibilities is not None:
Expand Down
194 changes: 194 additions & 0 deletions test/test_pebble.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -442,6 +442,59 @@ def test_file_info_from_dict(self):
self.assertEqual(info.group_id, 34)
self.assertEqual(info.group, 'staff')

def test_notice_from_dict(self):
notice = pebble.Notice.from_dict({
'id': '123',
'user-id': 1000,
'type': 'custom',
'key': 'example.com/a',
'visibility': 'private',
'first-occurred': '2023-12-07T17:01:02.123456789Z',
'last-occurred': '2023-12-07T17:01:03.123456789Z',
'last-repeated': '2023-12-07T17:01:04.123456789Z',
'occurrences': 7,
'last-data': {'k1': 'v1', 'k2': 'v2'},
'repeat-after': '30m',
'expire-after': '24h',
})
self.assertEqual(notice, pebble.Notice(
id='123',
user_id=1000,
type=pebble.NoticeType.CUSTOM,
key='example.com/a',
visibility=pebble.NoticeVisibility.PRIVATE,
first_occurred=datetime_utc(2023, 12, 7, 17, 1, 2, 123457),
last_occurred=datetime_utc(2023, 12, 7, 17, 1, 3, 123457),
last_repeated=datetime_utc(2023, 12, 7, 17, 1, 4, 123457),
occurrences=7,
last_data={'k1': 'v1', 'k2': 'v2'},
repeat_after=datetime.timedelta(minutes=30),
expire_after=datetime.timedelta(hours=24),
))

notice = pebble.Notice.from_dict({
'id': '124',
'user-id': 1001,
'type': 'other',
'key': 'example.com/b',
'visibility': 'public',
'first-occurred': '2023-12-07T17:01:02.123456789Z',
'last-occurred': '2023-12-07T17:01:03.123456789Z',
'last-repeated': '2023-12-07T17:01:04.123456789Z',
'occurrences': 8,
})
self.assertEqual(notice, pebble.Notice(
id='124',
user_id=1001,
type='other',
key='example.com/b',
visibility=pebble.NoticeVisibility.PUBLIC,
first_occurred=datetime_utc(2023, 12, 7, 17, 1, 2, 123457),
last_occurred=datetime_utc(2023, 12, 7, 17, 1, 3, 123457),
last_repeated=datetime_utc(2023, 12, 7, 17, 1, 4, 123457),
occurrences=8,
))


class TestPlan(unittest.TestCase):
def test_no_args(self):
Expand Down Expand Up @@ -2702,6 +2755,147 @@ def test_checklevel_conversion(self):
('GET', '/v1/checks', {'level': 'ready', 'names': ['chk2']}, None),
])

def test_get_notice(self):
self.client.responses.append({
'result': {
'id': '123',
'user-id': 1000,
'type': 'custom',
'key': 'example.com/a',
'visibility': 'private',
'first-occurred': '2023-12-07T17:01:02.123456789Z',
'last-occurred': '2023-12-07T17:01:03.123456789Z',
'last-repeated': '2023-12-07T17:01:04.123456789Z',
'occurrences': 7,
},
'status': 'OK',
'status-code': 200,
'type': 'sync',
})

notice = self.client.get_notice('124')

# No need to re-test full Notice.from_dict behaviour
self.assertEqual(notice.id, '123')

self.assertEqual(self.client.requests, [
('GET', '/v1/notices/124', None, None),
])

def test_get_notice_not_found(self):
self.client.responses.append(pebble.APIError({}, 404, 'Not Found', 'not found'))

with self.assertRaises(pebble.APIError) as cm:
self.client.get_notice('1')
self.assertEqual(cm.exception.code, 404)

self.assertEqual(self.client.requests, [
('GET', '/v1/notices/1', None, None),
])

def test_get_notices_all(self):
self.client.responses.append({
'result': [{
'id': '123',
'user-id': 1000,
'type': 'custom',
'key': 'example.com/a',
'visibility': 'private',
'first-occurred': '2023-12-07T17:01:02.123456789Z',
'last-occurred': '2023-12-07T17:01:03.123456789Z',
'last-repeated': '2023-12-07T17:01:04.123456789Z',
'occurrences': 7,
}, {
'id': '124',
'user-id': 1001,
'type': 'other',
'key': 'example.com/b',
'visibility': 'public',
'first-occurred': '2023-12-07T17:01:02.123456789Z',
'last-occurred': '2023-12-07T17:01:03.123456789Z',
'last-repeated': '2023-12-07T17:01:04.123456789Z',
'occurrences': 8,
}],
'status': 'OK',
'status-code': 200,
'type': 'sync',
})

checks = self.client.get_notices()
self.assertEqual(len(checks), 2)
self.assertEqual(checks[0].id, '123')
self.assertEqual(checks[1].id, '124')

self.assertEqual(self.client.requests, [
('GET', '/v1/notices', {}, None),
])

def test_get_notices_filters(self):
self.client.responses.append({
'result': [{
'id': '123',
'user-id': 1000,
'type': 'custom',
'key': 'example.com/a',
'visibility': 'private',
'first-occurred': '2023-12-07T17:01:02.123456789Z',
'last-occurred': '2023-12-07T17:01:03.123456789Z',
'last-repeated': '2023-12-07T17:01:04.123456789Z',
'occurrences': 7,
}, {
'id': '124',
'user-id': 1001,
'type': 'other',
'key': 'example.com/b',
'visibility': 'public',
'first-occurred': '2023-12-07T17:01:02.123456789Z',
'last-occurred': '2023-12-07T17:01:03.123456789Z',
'last-repeated': '2023-12-07T17:01:04.123456789Z',
'occurrences': 8,
}],
'status': 'OK',
'status-code': 200,
'type': 'sync',
})

notices = self.client.get_notices(
user_ids=[1000, 1001],
special_user=pebble.NoticeSpecialUser.SELF,
types=[pebble.NoticeType.CUSTOM],
keys=['example.com/a', 'example.com/b'],
visibilities=[pebble.NoticeVisibility.PRIVATE, pebble.NoticeVisibility.PUBLIC],
after=datetime_utc(2023, 12, 1, 2, 3, 4, 5),
)
self.assertEqual(len(notices), 2)
self.assertEqual(notices[0].id, '123')
self.assertEqual(notices[1].id, '124')

query = {
'user-ids': ['1000', '1001', 'self'],
'types': ['custom'],
'keys': ['example.com/a', 'example.com/b'],
'visibilities': ['private', 'public'],
'after': '2023-12-01T02:03:04.000005+00:00',
}
self.assertEqual(self.client.requests, [
('GET', '/v1/notices', query, None),
])

def test_get_notices_special_user_only(self):
self.client.responses.append({
'result': [],
'status': 'OK',
'status-code': 200,
'type': 'sync',
})

notices = self.client.get_notices(special_user=pebble.NoticeSpecialUser.ALL)
self.assertEqual(notices, [])

self.assertEqual(self.client.requests, [
('GET', '/v1/notices', {'user-ids': ['all']}, None),
])


class TestSocketClient(unittest.TestCase):
def test_socket_not_found(self):
Expand Down

0 comments on commit a1eb569

Please sign in to comment.