pytchat-fork/tests/test_compatible_processor.py

import json
from pytchat.parser.live import Parser
from pytchat.processors.compatible.processor import CompatibleProcessor

parser = Parser(is_replay=False)


def test_textmessage(mocker):
    '''api互換processorのテスト：通常テキストメッセージ'''
    processor = CompatibleProcessor()

    _json = _open_file("tests/testdata/compatible/textmessage.json")

    _, chatdata = parser.parse(parser.get_contents(json.loads(_json))[0])
    data = {
        "video_id": "",
        "timeout": 7,
        "chatdata": chatdata
    }
    ret = processor.process([data])

    assert ret["kind"] == "youtube#liveChatMessageListResponse"
    assert ret["pollingIntervalMillis"] == data["timeout"] * 1000
    assert ret.keys() == {
        "kind", "etag", "pageInfo", "nextPageToken", "pollingIntervalMillis", "items"
    }
    assert ret["pageInfo"].keys() == {
        "totalResults", "resultsPerPage"
    }
    assert ret["items"][0].keys() == {
        "kind", "etag", "id", "snippet", "authorDetails"
    }
    assert ret["items"][0]["snippet"].keys() == {
        'type', 'liveChatId', 'authorChannelId', 'publishedAt', 'hasDisplayContent', 'displayMessage',
        'textMessageDetails'
    }
    assert ret["items"][0]["authorDetails"].keys() == {
        'channelId', 'channelUrl', 'displayName', 'profileImageUrl', 'isVerified', 'isChatOwner', 'isChatSponsor',
        'isChatModerator'
    }
    assert ret["items"][0]["snippet"]["textMessageDetails"].keys() == {
        'messageText'
    }
    assert "LCC." in ret["items"][0]["id"]
    assert ret["items"][0]["snippet"]["type"] == "textMessageEvent"


def test_newsponcer(mocker):
    '''api互換processorのテスト：メンバ新規登録'''
    processor = CompatibleProcessor()

    _json = _open_file("tests/testdata/compatible/newSponsor.json")

    _, chatdata = parser.parse(parser.get_contents(json.loads(_json))[0])
    data = {
        "video_id": "",
        "timeout": 7,
        "chatdata": chatdata
    }
    ret = processor.process([data])

    assert ret["kind"] == "youtube#liveChatMessageListResponse"
    assert ret["pollingIntervalMillis"] == data["timeout"] * 1000
    assert ret.keys() == {
        "kind", "etag", "pageInfo", "nextPageToken", "pollingIntervalMillis", "items"
    }
    assert ret["pageInfo"].keys() == {
        "totalResults", "resultsPerPage"
    }
    assert ret["items"][0].keys() == {
        "kind", "etag", "id", "snippet", "authorDetails"
    }
    assert ret["items"][0]["snippet"].keys() == {
        'type', 'liveChatId', 'authorChannelId', 'publishedAt', 'hasDisplayContent', 'displayMessage'

    }
    assert ret["items"][0]["authorDetails"].keys() == {
        'channelId', 'channelUrl', 'displayName', 'profileImageUrl', 'isVerified', 'isChatOwner', 'isChatSponsor',
        'isChatModerator'
    }
    assert "LCC." in ret["items"][0]["id"]
    assert ret["items"][0]["snippet"]["type"] == "newSponsorEvent"


def test_newsponcer_rev(mocker):
    '''api互換processorのテスト：メンバ新規登録'''
    processor = CompatibleProcessor()

    _json = _open_file("tests/testdata/compatible/newSponsor_rev.json")

    _, chatdata = parser.parse(parser.get_contents(json.loads(_json))[0])
    data = {
        "video_id": "",
        "timeout": 7,
        "chatdata": chatdata
    }
    ret = processor.process([data])

    assert ret["kind"] == "youtube#liveChatMessageListResponse"
    assert ret["pollingIntervalMillis"] == data["timeout"] * 1000
    assert ret.keys() == {
        "kind", "etag", "pageInfo", "nextPageToken", "pollingIntervalMillis", "items"
    }
    assert ret["pageInfo"].keys() == {
        "totalResults", "resultsPerPage"
    }
    assert ret["items"][0].keys() == {
        "kind", "etag", "id", "snippet", "authorDetails"
    }
    assert ret["items"][0]["snippet"].keys() == {
        'type', 'liveChatId', 'authorChannelId', 'publishedAt', 'hasDisplayContent', 'displayMessage'

    }
    assert ret["items"][0]["authorDetails"].keys() == {
        'channelId', 'channelUrl', 'displayName', 'profileImageUrl', 'isVerified', 'isChatOwner', 'isChatSponsor',
        'isChatModerator'
    }
    assert "LCC." in ret["items"][0]["id"]
    assert ret["items"][0]["snippet"]["type"] == "newSponsorEvent"


def test_superchat(mocker):
    '''api互換processorのテスト：スパチャ'''
    processor = CompatibleProcessor()

    _json = _open_file("tests/testdata/compatible/superchat.json")

    _, chatdata = parser.parse(parser.get_contents(json.loads(_json))[0])
    data = {
        "video_id": "",
        "timeout": 7,
        "chatdata": chatdata
    }
    ret = processor.process([data])

    assert ret["kind"] == "youtube#liveChatMessageListResponse"
    assert ret["pollingIntervalMillis"] == data["timeout"] * 1000
    assert ret.keys() == {
        "kind", "etag", "pageInfo", "nextPageToken", "pollingIntervalMillis", "items"
    }
    assert ret["pageInfo"].keys() == {
        "totalResults", "resultsPerPage"
    }
    assert ret["items"][0].keys() == {
        "kind", "etag", "id", "snippet", "authorDetails"
    }
    assert ret["items"][0]["snippet"].keys() == {
        'type', 'liveChatId', 'authorChannelId', 'publishedAt', 'hasDisplayContent', 'displayMessage',
        'superChatDetails'
    }
    assert ret["items"][0]["authorDetails"].keys() == {
        'channelId', 'channelUrl', 'displayName', 'profileImageUrl', 'isVerified', 'isChatOwner', 'isChatSponsor',
        'isChatModerator'
    }
    assert ret["items"][0]["snippet"]["superChatDetails"].keys() == {
        'amountMicros', 'currency', 'amountDisplayString', 'tier', 'backgroundColor'
    }
    assert "LCC." in ret["items"][0]["id"]
    assert ret["items"][0]["snippet"]["type"] == "superChatEvent"


def test_unregistered_currency(mocker):
    processor = CompatibleProcessor()

    _json = _open_file("tests/testdata/unregistered_currency.json")

    _, chatdata = parser.parse(parser.get_contents(json.loads(_json))[0])

    data = {
        "video_id": "",
        "timeout": 7,
        "chatdata": chatdata
    }
    ret = processor.process([data])
    assert ret["items"][0]["snippet"]["superChatDetails"]["currency"] == "[UNREGISTERD]"


def _open_file(path):
    with open(path, mode='r', encoding='utf-8') as f:
        return f.read()