import unittest
import atexit
import os
from multiprocessing.pool import Pool
import asyncio
import aiohttp
import requests
import re
from itertools import *
import faker
from faker.providers import internet
from faker.providers import lorem
from faker.providers import misc
from src.settings import config
from src.server import app
from test.testcase import AsyncioTestCase
fake = faker.Faker()
fake.add_provider(internet)
fake.add_provider(lorem)
fake.add_provider(misc)
def fake_lbryusername():
return '@' + fake.user_name()
async def jsonrpc_post(url, method, **params):
json_body = {
'jsonrpc': '2.0',
'id': None,
'method': method,
'params': params
}
async with aiohttp.request('POST', url, json=json_body) as request:
return await request.json()
def nothing():
pass
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]
class ServerTest(AsyncioTestCase):
db_file = 'test.db'
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.url = 'http://' + config['HOST'] + ':5921/api'
@classmethod
def tearDownClass(cls) -> None:
print('exit reached')
os.remove(cls.db_file)
async def asyncSetUp(self):
await super().asyncSetUp()
self.server = app.CommentDaemon(config, db_file=self.db_file)
await self.server.start()
self.addCleanup(self.server.stop)
async def post_comment(self, **params):
return await jsonrpc_post(self.url, 'create_comment', **params)
def is_valid_message(self, comment=None, claim_id=None, parent_id=None,
channel_name=None, channel_id=None, signature=None, signing_ts=None):
try:
assert comment is not None and claim_id is not None
assert re.fullmatch('([a-f0-9]|[A-F0-9]){40}', claim_id)
assert 0 < len(comment) <= 2000
if parent_id is not None:
assert re.fullmatch('([a-f0-9]){64}', parent_id)
if channel_name or channel_id or signature or signing_ts:
assert channel_id is not None and channel_name is not None
assert re.fullmatch('([a-f0-9]|[A-F0-9]){40}', channel_id)
assert self.valid_channel_name(channel_name)
assert (signature is None and signing_ts is None) or \
(signature is not None and signing_ts is not None)
if signature:
assert len(signature) == 128
if parent_id:
assert parent_id.isalnum()
except Exception:
return False
return True
@staticmethod
def valid_channel_name(channel_name):
return re.fullmatch(
'^@(?:(?![\x00-\x08\x0b\x0c\x0e-\x1f\x23-\x26'
'\x2f\x3a\x3d\x3f-\x40\uFFFE-\U0000FFFF]).){1,255}$',
channel_name
)
async 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 = await self.post_comment(**test)
self.assertTrue('result' in message or 'error' in message)
if 'error' in message:
self.assertFalse(self.is_valid_message(**test))
else:
self.assertTrue(self.is_valid_message(**test))
async 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 = await self.post_comment(**test)
self.assertTrue('result' in message or 'error' in message)
if 'error' in message:
self.assertFalse(self.is_valid_message(**test))
else:
self.assertTrue(self.is_valid_message(**test))
async 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 = await self.post_comment(**test)
self.assertTrue('result' in message or 'error' in message)
if 'error' in message:
self.assertFalse(self.is_valid_message(**test))
else:
self.assertTrue(self.is_valid_message(**test))
async def test04CreateAllReplies(self):
claim_id = '1d8a5cc39ca02e55782d619e67131c0a20843be8'
parent_comment = await 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['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):
if test['parent_id'] != parent_id:
continue
else:
message = await self.post_comment(**test)
self.assertTrue('result' in message or 'error' in message)
if 'error' in message:
self.assertFalse(self.is_valid_message(**test))
else:
self.assertTrue(self.is_valid_message(**test))
class ListCommentsTest(AsyncioTestCase):
replace = {
'claim_id': fake.sha1,
'comment': fake.text,
'channel_id': fake.sha1,
'channel_name': fake_lbryusername,
'signature': nothing,
'parent_id': nothing
}
db_file = 'list_test.db'
url = 'http://localhost:5921/api'
comment_ids = None
claim_id = '1d8a5cc39ca02e55782d619e67131c0a20843be8'
@classmethod
async def post_comment(cls, **params):
return await jsonrpc_post(cls.url, 'create_comment', **params)
@classmethod
def tearDownClass(cls) -> None:
print('exit reached')
os.remove(cls.db_file)
async def asyncSetUp(self):
await super().asyncSetUp()
self.server = app.CommentDaemon(config, db_file=self.db_file)
await self.server.start()
self.addCleanup(self.server.stop)
if self.comment_ids is None:
self.comment_list = [{key: self.replace[key]() for key in self.replace.keys()} for _ in range(23)]
for comment in self.comment_list:
comment['claim_id'] = self.claim_id
self.comment_ids = [(await self.post_comment(**comm))['result']['comment_id']
for comm in self.comment_list]
async def testListComments(self):
response_one = await jsonrpc_post(self.url, 'get_claim_comments', page_size=20,
page=1, top_level=1, claim_id=self.claim_id)
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'], len(response_one['items']))
self.assertIn('items', response_one)
self.assertGreaterEqual(response_one['total_pages'], response_one['page'])
last_page = response_one['total_pages']
response = await jsonrpc_post(self.url, 'get_claim_comments', page_size=20,
page=last_page, top_level=1, claim_id=self.claim_id)
self.assertIsNotNone(response)
self.assertIn('result', response)
response: dict = response['result']
self.assertIs(type(response['items']), list)
self.assertEqual(response['total_items'], response_one['total_items'])
self.assertEqual(response['total_pages'], response_one['total_pages'])