1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
|
from typing import Any
from mongomock import MongoClient
from pymongo.collection import Collection
from schedule import Scheduler
from unittest import TestCase
from components.subscriptions.main import Subscription
from components.subscriptions.typing import SubsDict
from components.users.typing import UserDict
class TestFeeds(TestCase):
def setUp(self) -> None:
self.client: MongoClient[Any] = MongoClient(tz_aware=True)
self.collection: Collection[SubsDict] = self.client.db.collection
self.scheduler: Scheduler = Scheduler()
def test_insert(self) -> None:
sub = Subscription(
_id="yt:channel:bla",
link="http://www.youtube.com/feeds/videos.xml?channel_id=UCbla",
time_between_fetches=5,
)
sub._collection = self.collection
sub._scheduler = self.scheduler
sub.insert()
sub_dict = self.collection.find_one({"_id": "yt:channel:bla"})
self.assertIsNotNone(sub_dict)
assert sub_dict # To appease mypy.
self.assertDictEqual(sub.asdict(), sub_dict)
def test_feed_fetch(self) -> None:
sub = Subscription(
_id="yt:channel:hlgI3UHCOnwUGzWzbJ3H5w",
link=r"tests/data/feed@ytnnews24@001.xml",
time_between_fetches=1,
)
sub._collection = self.collection
sub._scheduler = self.scheduler
sub.insert()
sub.initialise_job()
self.scheduler.run_all()
self.assertEqual(15, len(sub.videos))
sub_dict = self.collection.find_one({"_id": "yt:channel:hlgI3UHCOnwUGzWzbJ3H5w"})
self.assertIsNotNone(sub_dict)
assert sub_dict # To appease mypy.
self.assertEqual(15, len(sub_dict["videos"]))
def test_feed_update(self) -> None:
sub = Subscription(
_id="yt:channel:hlgI3UHCOnwUGzWzbJ3H5w",
link=r"tests/data/feed@ytnnews24@001.xml",
time_between_fetches=1,
)
sub._collection = self.collection
sub._scheduler = self.scheduler
sub.insert()
sub.initialise_job()
self.scheduler.run_all()
sub.link=r"tests/data/feed@ytnnews24@002.xml"
self.scheduler.run_all()
self.assertEqual(16, len(sub.videos))
sub_dict = self.collection.find_one({"_id": "yt:channel:hlgI3UHCOnwUGzWzbJ3H5w"})
self.assertIsNotNone(sub_dict)
assert sub_dict # To appease mypy.
self.assertEqual(16, len(sub_dict["videos"]))
def tearDown(self) -> None:
self.client.close()
|
