Add tests

tests/test_calculator_get_item.py

@@ -1,140 +0,0 @@
-from pytchat.processors.superchat.calculator import SuperchatCalculator
-
-get_item = SuperchatCalculator()._get_item
-
-dict_test = {
-    'root':{
-        'node0' : 'value0',
-        'node1' : 'value1',
-        'node2' : {
-            'node2-0' : 'value2-0'
-        },
-    
-        'node3' : [
-            {'node3-0' : 'value3-0'},
-            {'node3-1' : 
-                {'node3-1-0' : 'value3-1-0'}
-            }
-        ],
-        'node4' : [],
-        'node5' : [
-            [
-                {'node5-1-0' : 'value5-1-0'},
-                {'node5-1-1' : 'value5-1-1'},
-            ],
-                {'node5-0' : 'value5-0'},
-            
-        ]
-    }
-}
-
-items_test0 = [
-    'root',
-    'node1'
-]
-
-
-items_test_not_found0 = [
-    'root',
-    'other_data'
-]
-
-
-items_test_nest = [
-    'root',
-    'node2',
-    'node2-0'
-]
-
-items_test_list0 = [
-    'root',
-    'node3',
-    1,
-    'node3-1'
-]
-
-items_test_list1 = [
-    'root',
-    'node3',
-    1,
-    'node3-1',
-    'node3-1-0'
-]
-
-items_test_list2 = [
-    'root',
-    'node4',
-    None
-]
-
-items_test_list3 = [
-    'root',
-    'node4'
-]
-
-items_test_list_nest = [
-    'root',
-    'node5',
-    0,
-    1,
-    'node5-1-1'
-]
-
-items_test_list_nest_not_found1 = [
-    'root',
-    'node5',
-    0,
-    1,
-    'node5-1-1',
-    'nodez'
-]
-
-items_test_not_found1 = [
-    'root',
-    'node3',
-    2,
-    'node3-1',
-    'node3-1-0'
-]
-
-items_test_not_found2 = [
-    'root',
-    'node3',
-    2,
-    'node3-1',
-    'node3-1-0',
-    'nodex'
-]
-def test_get_items_0():
-    assert get_item(dict_test, items_test0) == 'value1'
-
-def test_get_items_1():
-    assert get_item(dict_test, items_test_not_found0) is None
-
-def test_get_items_2():
-    assert get_item(dict_test, items_test_nest) == 'value2-0'
-
-def test_get_items_3():
-    assert get_item(
-        dict_test, items_test_list0) ==  {'node3-1-0' : 'value3-1-0'}
-
-def test_get_items_4():
-    assert get_item(dict_test, items_test_list1) ==  'value3-1-0'
-
-def test_get_items_5():
-    assert get_item(dict_test, items_test_not_found1) ==  None
-
-def test_get_items_6():
-    assert get_item(dict_test, items_test_not_found2) ==  None
-
-def test_get_items_7():
-    assert get_item(dict_test, items_test_list2) ==  None
-
-def test_get_items_8():
-    assert get_item(dict_test, items_test_list_nest) ==  'value5-1-1'
-
-def test_get_items_9():
-    assert get_item(dict_test, items_test_list_nest_not_found1) ==  None
-
-def test_get_items_10():
-    assert get_item(dict_test, items_test_list3) ==  []

tests/test_calculator_parse.py

@@ -67,6 +67,5 @@ def test_process_2():
             'chatdata': load_chatdata(r"tests/testdata/calculator/replay_end.json")
         }
         assert False
-        SuperchatCalculator().process([chat_component])
     except ChatParseException:
         assert True

tests/test_compatible_processor2.py

@@ -0,0 +1,24 @@
+import pytchat
+from pytchat.processors.compatible.processor import CompatibleProcessor
+
+
+root_keys = ('kind', 'etag', 'nextPageToken', 'pollingIntervalMillis', 'pageInfo', 'items')
+item_keys = ('kind', 'etag', 'id', 'snippet', 'authorDetails')
+snippet_keys = ('type', 'liveChatId', 'authorChannelId', 'publishedAt', 'hasDisplayContent', 'displayMessage', 'textMessageDetails')
+author_details_keys = ('channelId', 'channelUrl', 'displayName', 'profileImageUrl', 'isVerified', 'isChatOwner', 'isChatSponsor', 'isChatModerator')
+
+def test_compatible_processor():
+    stream = pytchat.create("Hj-wnLIYKjw", seektime = 6000, processor=CompatibleProcessor())
+    while stream.is_alive():
+        chat = stream.get()
+        for key in chat.keys():
+            assert key in root_keys
+        for key in chat["items"][0].keys():
+            assert key in item_keys
+        for key in chat["items"][0]["snippet"].keys():
+            assert key in snippet_keys
+        for key in chat["items"][0]["authorDetails"].keys():
+            assert key in author_details_keys
+        break
+
+    

tests/test_default_processor2.py

@@ -0,0 +1,94 @@
+import asyncio
+import pytchat
+from concurrent.futures import CancelledError
+from pytchat.core_multithread.livechat import LiveChat
+from pytchat.core_async.livechat import LiveChatAsync
+
+cases = [
+    {
+        "video_id":"1X7oL0hDnMg", "seektime":1620,
+        "result1":{'textMessage': 84},
+        "result2":{'': 83, 'MODERATOR': 1}
+    },
+    {
+        "video_id":"Hj-wnLIYKjw", "seektime":420,
+        "result1":{'superChat': 1, 'newSponsor': 6, 'textMessage': 63, 'donation': 1},
+        "result2":{'': 59, 'MEMBER': 12}
+    },
+    {
+        "video_id":"S8dmq5YIUoc", "seektime":3,
+        "result1":{'textMessage': 86},
+        "result2":{'': 62, 'MEMBER': 21, 'OWNER': 2, 'VERIFIED': 1}
+    },{
+        "video_id":"yLrstz80MKs", "seektime":30,
+        "result1":{'superSticker': 8, 'superChat': 2, 'textMessage': 67},
+        "result2":{'': 73, 'MEMBER': 4}
+    }
+]
+
+def test_archived_stream():
+    for case in cases:
+        stream = pytchat.create(video_id=case["video_id"], seektime=case["seektime"])
+        while stream.is_alive():
+            chat = stream.get()
+            agg1 = {}
+            agg2 = {}
+            for c in chat.items:
+                if c.type in agg1:
+                    agg1[c.type] += 1
+                else:
+                    agg1[c.type] = 1
+                if c.author.type in agg2:
+                    agg2[c.author.type] += 1
+                else:
+                    agg2[c.author.type] = 1
+            break
+        assert agg1 == case["result1"]
+        assert agg2 == case["result2"]
+
+
+def test_archived_stream_multithread():
+    for case in cases:
+        stream = LiveChat(video_id=case["video_id"], seektime=case["seektime"])
+        while stream.is_alive():
+            chat = stream.get()
+            agg1 = {}
+            agg2 = {}
+            for c in chat.items:
+                if c.type in agg1:
+                    agg1[c.type] += 1
+                else:
+                    agg1[c.type] = 1
+                if c.author.type in agg2:
+                    agg2[c.author.type] += 1
+                else:
+                    agg2[c.author.type] = 1
+            break
+        assert agg1 == case["result1"]
+        assert agg2 == case["result2"]
+    
+def test_async_live_stream():
+    async def test_loop(): 
+        for case in cases:
+            stream = LiveChatAsync(video_id=case["video_id"], seektime=case["seektime"])
+            while stream.is_alive():
+                chat = await stream.get()
+                agg1 = {}
+                agg2 = {}
+                for c in chat.items:
+                    if c.type in agg1:
+                        agg1[c.type] += 1
+                    else:
+                        agg1[c.type] = 1
+                    if c.author.type in agg2:
+                        agg2[c.author.type] += 1
+                    else:
+                        agg2[c.author.type] = 1
+                break
+            assert agg1 == case["result1"]
+            assert agg2 == case["result2"]
+    loop = asyncio.get_event_loop()
+    try:
+        loop.run_until_complete(test_loop())
+    except CancelledError:
+        assert True

tests/test_speed_calculator2.py

@@ -0,0 +1,16 @@
+import json
+import pytchat
+from pytchat.parser.live import Parser
+from pytchat.processors.speed.calculator import SpeedCalculator
+
+parser = Parser(is_replay=False)
+
+
+def test_speed_1():
+    stream = pytchat.create("Hj-wnLIYKjw", seektime = 6000,processor=SpeedCalculator())
+    while stream.is_alive():
+        speed = stream.get()
+        assert speed > 100
+        break
+test_speed_1()
+