diff --git a/tests/server_test.py b/tests/server_test.py index d7914f9..02fe848 100644 --- a/tests/server_test.py +++ b/tests/server_test.py @@ -1,10 +1,7 @@ import unittest -from random import randint -import requests -import re import faker -from itertools import * +import requests from faker.providers import internet from faker.providers import lorem from faker.providers import misc @@ -17,246 +14,122 @@ fake.add_provider(lorem) fake.add_provider(misc) -class ServerTest(unittest.TestCase): - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - self.url = 'http://' + config['HOST'] + ':5921/api' - - - def post_comment(self, **params): - json_body = { +def post_jsonrpc(url, method, **params): + return requests.post( + url=url, + json={ 'jsonrpc': '2.0', 'id': None, - 'method': 'create_comment', + 'method': method, 'params': params } - return requests.post(url=self.url, json=json_body) + ).json() - def assertIsValidMessageTest(self, message, test): - self.assertIsNotNone(message) - try: - if not test['claim_id'] or \ - (bool(test['channel_id']) ^ bool(test['channel_name'])): - self.assertIn('error', message) - self.assertNotIn('result', message) - else: - self.assertNotIn('error', message) - self.assertIn('result', message) - self.assertIn('comment_id', message['result']) - self.assertEquals(message['result']['claim_id'], test['claim_id']) - except AssertionError: - raise requests.HTTPError(message.text) - def isValidMessage(self, message: dict): - return message and type(message) is dict and ('error' in message or 'result' in message) +def comment_create(**kwargs): + return post_jsonrpc('http://localhost:5921/api', 'create_comment', **kwargs) - def isValidTest(self, test: dict): - cond = test['claim_id'] and test['comment'] and not \ - (bool(test['channel_id']) ^ bool(test['channel_name'])) - if cond: - cond = (0 < len(test['comment']) <= 2000) and cond - if test['channel_id']: - cond = (1 < len(test['channel_name']) <= 256) and cond - channel_match = re.fullmatch( - '^@(?:(?![\x00-\x08\x0b\x0c\x0e-\x1f\x23-\x26' - '\x2f\x3a\x3d\x3f-\x40\uFFFE-\U0000FFFF]).){1,255}$', - test['channel_name'] - ) - cond = cond and channel_match - return cond - def setUp(self) -> None: - self.reply_id = 'ace7800f36e55c74c4aa6a698f97a7ee5f1ccb047b5a0730960df90e58c41dc2' - - def test01CreateCommentNoReply(self): - anonymous_test = create_test_comments( - ('claim_id', 'channel_id', 'channel_name', 'comment'), - comment=None, - channel_name=None, - channel_id=None, - claim_id=None - ) - for test in anonymous_test: - with self.subTest(test=test): - message = self.post_comment(**test) - message = message.json() - if self.isValidTest(test): - self.assertIn('result', message) - self.assertIsNotNone(message['result']) - self.assertIn('comment_id', message['result']) - else: - self.assertIn('error', message) - - def test02CreateNamedCommentsNoReply(self): - named_test = create_test_comments( - ('channel_name', 'channel_id', 'signature'), - claim_id='1234567890123456789012345678901234567890', - channel_name='@JamieFoxx', - channel_id='1234567890'*4, - comment='blah blah blahbity blah', - signature=None - ) - for test in named_test: - with self.subTest(test=test): - message = self.post_comment(**test) - message = message.json() - if self.isValidTest(test): - self.assertTrue(self.isValidMessage(message)) - else: - self.assertFalse(self.isValidMessage(message)) - - def test03CreateAllTestComments(self): - test_all = create_test_comments(replace.keys(), **{ - k: None for k in replace.keys() - }) - for test in test_all: - with self.subTest(test=test): - message = self.post_comment(**test) - message = message.json() - if self.isValidTest(test): - self.assertTrue(self.isValidMessage(message)) - self.assertNotIn('error', message) - self.assertIsNotNone(message['result']) - else: - self.assertIsNotNone(message) - self.assertIn('error', message) - - def test04CreateAllReplies(self): - claim_id = '1d8a5cc39ca02e55782d619e67131c0a20843be8' - parent_comment = self.post_comment( - channel_name='@KevinWalterRabie', - channel_id=fake.sha1(), - comment='Hello everybody and welcome back to my chan nel', - claim_id=claim_id, - ) - parent_id = parent_comment.json()['result']['comment_id'] - test_all = create_test_comments( - ('comment', 'channel_name', 'channel_id', 'signature', 'parent_id'), - parent_id=parent_id, - comment=None, - channel_name=None, - channel_id=None, - signature=None, - claim_id=claim_id - ) - for test in test_all: - with self.subTest(test=test) as subtest: - if test['parent_id'] != parent_id: - continue - else: - message = self.post_comment(**test) - message = message.json() - if self.isValidTest(test): - self.assertTrue(self.isValidMessage(message)) - self.assertNotIn('error', message) - self.assertIsNotNone(message['result']) - message = message['result'] - self.assertIn('parent_id', message) - self.assertEquals(message['parent_id'], parent_id) - else: - self.assertIn('error', message) +def comment_list(**kwargs): + return post_jsonrpc('http://localhost:5921/api', 'get_claim_comments', **kwargs) def fake_lbryusername(): return '@' + fake.user_name() -def jsonrpc_post(url, method, **params): - json_body = { - 'jsonrpc': '2.0', - 'id': None, - 'method': method, - 'params': params - } - return requests.post(url=url, json=json_body) +class ServerTest(unittest.TestCase): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.url = 'http://' + config['HOST'] + ':5921/api' + def setUp(self) -> None: + self.claim_id = '9cb713f01bf247a0e03170b5ed00d5161340c486' -def nothing(): - return None + def test01CreateCommentNoReply(self): + resp = comment_create( + claim_id=self.claim_id, + comment='anonymous comment' + ) + self.assertIn('result', resp) + resp = resp['result'] + self.assertIsNotNone(resp) + self.assertIn('comment_id', resp) + def test02CreateNamedCommentNoSignature(self): + resp = comment_create( + claim_id=self.claim_id, + comment='blah blah blah this is my comment', + channel_name=fake_lbryusername(), + channel_id=fake.sha1(), + ) + self.assertIn('result', resp) + resp = resp['result'] + self.assertIsNotNone(resp) + self.assertIn('comment_id', resp) + self.assertNotIn('parent_id', resp) -class ListCommentsTest(unittest.TestCase): - replace = { - 'claim_id': fake.sha1, - 'comment': fake.text, - 'channel_id': fake.sha1, - 'channel_name': fake_lbryusername, - 'signature': nothing, - 'parent_id': nothing - } - - @classmethod - def post_comment(cls, **params): - json_body = { - 'jsonrpc': '2.0', - 'id': None, - 'method': 'create_comment', - 'params': params - } - return requests.post(url=cls.url, json=json_body) - - @classmethod - def setUpClass(cls) -> None: - cls.url = 'http://' + config['HOST'] + ':5921/api' - cls.claim_id = '1d8a5cc39ca02e55782d619e67131c0a20843be8' - cls.comment_list = [{key: cls.replace[key]() for key in cls.replace.keys()} for _ in range(23)] - for comment in cls.comment_list: - comment['claim_id'] = cls.claim_id - cls.comment_ids = [cls.post_comment(**comm).json()['result']['comment_id'] - for comm in cls.comment_list] - - def testListComments(self): - response_one = jsonrpc_post(self.url, 'get_claim_comments', page_size=20, - page=1, top_level=1, claim_id=self.claim_id).json() - self.assertIsNotNone(response_one) - self.assertIn('result', response_one) - response_one: dict = response_one['result'] - self.assertIs(type(response_one), dict) - self.assertEquals(response_one['page_size'], 20) - self.assertIn('items', response_one) - self.assertEqual(response_one['total_pages'], 2) - response = jsonrpc_post(self.url, 'get_claim_comments', page_size=20, - page=2, top_level=1, claim_id=self.claim_id).json() - self.assertIsNotNone(response) - self.assertIn('result', response) - response: dict = response['result'] - self.assertIs(response['items'], list) - self.assertEquals(len(response['items']), 3) - self.assertEqual(response['total_items'], response_one['total_items']) - self.assertEqual(response['total_pages'], 2) - result: list = response['items'] + response_one['items'] - for comment in result: - self.assertIsNotNone(comment) - self.assertIn('comment_id', comment) - self.assertIn(comment['comment_id'], self.comment_ids) - - - - -replace = { - 'claim_id': fake.sha1, - 'comment': fake.text, - 'channel_id': fake.sha1, - 'channel_name': fake_lbryusername, - 'signature': fake.uuid4, - 'parent_id': fake.sha256 -} - - -def create_test_comments(values: iter, **default): - vars_combo = chain.from_iterable(combinations(values, r) for r in range(1, len(values) + 1)) - return [{k: replace[k]() if k in comb else v for k, v in default.items()} - for comb in vars_combo] - - -def create_comment(channel_name, channel_id, claim_id=None, maxchar=500, reply_id=None, signature=None, parent_id=None): - return { - 'claim_id': claim_id if claim_id else fake.sha1(), - 'comment': ''.join(fake.text(max_nb_chars=maxchar)), - 'channel_name': channel_name, - 'channel_id': channel_id, - 'signature': signature if signature else fake.uuid4(), - 'parent_id': reply_id - } + def test03InvalidComments(self): + resp = comment_create( + claim_id=self.claim_id, + comment='this comment should fail', + signature=fake.sha1() + ) + self.assertNotIn('result', resp) + self.assertIn('error', resp) + resp = comment_create( + comment='This comment has no claim_id' + ) + self.assertNotIn('result', resp) + self.assertIn('error', resp) + resp = comment_create(claim_id=self.claim_id) + self.assertIn('error', resp) + resp = comment_create() + self.assertIn('error', resp) + resp = comment_create( + claim_id=self.claim_id, + comment='This comment has no channel_id', + channel_name=fake_lbryusername(), + ) + self.assertIn('error', resp) + resp = comment_create( + claim_id=self.claim_id, + comment='This comment has no channel_name', + channel_id=fake.sha1(), + ) + self.assertIn('error', resp) + def test04CreateAnonymousReply(self): + resp = comment_create(claim_id=self.claim_id, comment='anonymous comment') + self.assertIn('result', resp) + resp = resp['result'] + self.assertIn('comment_id', resp) + reply = comment_create( + claim_id=self.claim_id, + comment='anonymous reply', + parent_id=resp['comment_id'] + ) + self.assertIn('result', reply) + self.assertIn('parent_id', reply['result']) + self.assertEquals(reply['result']['parent_id'], resp['comment_id']) + def test05CreateNamedReplies(self): + resp = comment_create( + claim_id=self.claim_id, + comment='blah blah blah this is my comment', + channel_name=fake_lbryusername(), + channel_id=fake.sha1(), + ) + self.assertIn('result', resp) + parent_id = resp['result']['comment_id'] + reply = comment_create( + claim_id=self.claim_id, + comment='this is a boring response to a named comment', + channel_name='@DukeSilver', + channel_id=fake.sha1(), + parent_id=parent_id + ) + self.assertIn('result', reply) + self.assertIn('comment_id', reply['result']) + self.assertIn('parent_id', reply['result']) + self.assertEquals(reply['result']['parent_id'], parent_id)