/
tests.py
166 lines (134 loc) · 4.57 KB
/
tests.py
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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
import asyncio
import re
from datetime import datetime
import apraw
import async_subscan
async def test_subscan(reddit):
print("Async")
time_started = datetime.now()
async for sub in async_subscan.get_eligible():
if await sub.is_dead():
print(sub)
print(datetime.now() - time_started)
'''
print("Normal")
for sub in subscan.get_eligible():
print(sub)
'''
async def test_reddit(reddit):
# print(await reddit.get_request_headers())
# print(await reddit.message("dan6erbond", "test", "test"))
# print(await reddit.message("/r/jealousasfuck", "test", "test"))
print(await reddit.submission("db8k9e"))
await reddit.submission(
url="https://www.reddit.com/r/RandomActsOfGaming/comments/db8k9e/uplay_ghost_recon_wildlands")
async def test_redditor(reddit):
redditor = await reddit.redditor("dan6erbond")
# print(await redditor.message("test", "test"))
# async for sub in redditor.moderated_subreddits():
# print(sub)
async for c in redditor.comments():
s = await c.submission()
print(await s.subreddit())
async for s in redditor.submissions():
print(await s.subreddit())
async def test_submission(reddit):
subreddit = await reddit.subreddit("jealousasfuck")
async for s in subreddit.new(1):
print(await s.author())
print(await s.subreddit())
s = await reddit.submission("db8k9e")
print(s.title)
i = 0
parent_i = 0
ids = set()
async for c in s.comments():
if c.link_id != s.name:
print(c)
if c.parent_id == s.name:
parent_i += 1
i += 1
ids.add(c.id)
print("Comments found:", i)
print("Unique comments:", len(ids))
print("Parent-level comments:", parent_i)
async def test_comment(reddit):
subreddit = await reddit.subreddit("jealousasfuck")
async for s in subreddit.new(1):
async for c in s.comments():
print(await c.author())
c._submission = None
sub = await c.submission()
print(sub.id == s.id)
break
async def test_subreddit(reddit):
# subreddit = await reddit.subreddit("test")
# print(await subreddit.message("test", "test"))
subreddit = await reddit.subreddit("jealousasfuck")
ids = list()
duplicates = False
async for s in subreddit.new(None):
if s.id in ids:
duplicates = True
break
# print(type(s))
ids.append(s.id)
if len(ids) > 0 and not duplicates:
print("Test passed, {} posts returned.".format(len(ids)))
else:
print("Test failed, {} posts returned.".format(len(ids)))
ids = list()
async for ma in subreddit.mod.log():
ids.append(ma.id)
print("{} mod actions returned.".format(len(ids)))
return
subreddit = await reddit.subreddit("askreddit")
ids = list()
async for s in subreddit.new.stream():
print(s.id)
if s.id in ids:
print("Duplicate found:", s.id)
print(len(ids), " submissions found.")
break
ids.append(s.id)
# async for mod in subreddit.moderators():
# print(dir(await mod.redditor()))
async def test_subreddit_moderation(reddit):
subreddit = await reddit.subreddit("jealousasfuck")
async for i in subreddit.mod.reports():
print(i.user_reports)
print(i.mod_reports)
async def test_subreddits(reddit):
ids = list()
async for s in reddit.subreddits.new(limit=200):
ids.append(s.id)
print("{}:".format(len(ids)), s, "Subscribers:", s.subscribers)
async def test_modmail(reddit):
subreddit = await reddit.subreddit("jealousasfuck")
async for c in subreddit.modmail.conversations():
async for m in c.messages():
print(m.body_md)
break
break
tests = [test_subscan, test_reddit, test_redditor, test_subreddit, test_subreddit_moderation, test_subreddits,
test_redditor, test_submission, test_comment, test_modmail]
async def run_tests():
reddit = apraw.Reddit("D6B")
i = 0
for test in tests:
print("{}:".format(i), test)
i += 1
while True:
i = input("Select a test to run: ")
try:
i = int(i)
if i >= 0 and i < len(tests):
await tests[i](reddit)
else:
print("No test found at position {}!".format(i))
except Exception as e:
s = re.search("^(\d+)$", i)
if s is None:
break
loop = asyncio.get_event_loop()
loop.run_until_complete(run_tests())