comment-server/tests/server_test.py

import unittest
from random import randint

import requests
import re
import faker
from itertools import *
from faker.providers import internet
from faker.providers import lorem
from faker.providers import misc

from src.settings import config

fake = faker.Faker()
fake.add_provider(internet)
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 = {
            'jsonrpc': '2.0',
            'id': None,
            'method': 'create_comment',
            'params': params
        }
        return requests.post(url=self.url, json=json_body)

    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 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 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)


def nothing():
    return None


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
    }