-
Notifications
You must be signed in to change notification settings - Fork 14
/
dead_test.py
168 lines (124 loc) · 5.28 KB
/
dead_test.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
167
168
from unittest.mock import patch, ANY
from deadWeb.dead import DropHandler
import pprint
import datetime
import hashlib
from freezegun import freeze_time
@patch('pymongo.MongoClient')
@freeze_time("2012-01-14")
def test_track_is_saved(mock_pymongo):
dead = DropHandler(mock_pymongo)
dead.setRequestHash("127.0.0.1")
data = {"test":"here"}
dead.drop(data)
mock_pymongo.dead.track.insert_one.assert_called_with({"key": ANY,"userHash":ANY, "createdDate":datetime.datetime(2012, 1, 14),"pickedUp":ANY})
@patch('pymongo.MongoClient')
@freeze_time("2012-01-14")
def test_user_hash_is_truncated(mock_pymongo):
dead = DropHandler(mock_pymongo)
ipAddr ="127.0.0.1"
dead.setRequestHash(ipAddr)
data = {"test":"here"}
dead.drop(data)
saltedIP = DropHandler.salt +ipAddr
m = hashlib.sha256()
m.update(saltedIP.encode('utf-8'))
client_hash = m.hexdigest()[:32]
mock_pymongo.dead.track.insert_one.assert_called_with({"key": ANY,"userHash": client_hash, "createdDate":datetime.datetime(2012, 1, 14),"pickedUp":ANY})
@patch('pymongo.MongoClient')
@freeze_time("2012-01-14")
def test_drop_is_saved(mock_pymongo):
dead = DropHandler(mock_pymongo)
data = {"test":"here"}
dead.drop(data)
mock_pymongo.dead.drop.insert_one.assert_called_with({"key": ANY, "data":data,"createdDate":datetime.datetime(2012, 1, 14)})
def test_dummy_test():
assert True==True
@patch('pymongo.MongoClient')
@freeze_time("2012-01-14")
def test_drop_deleted_when_accessed(mock_pymongo):
sampleDrop = get_sample_drop()
mock_pymongo.dead.drop.find_one_and_delete.return_value=sampleDrop
dead = DropHandler(mock_pymongo)
import pprint
pprint.pprint("XXXXX")
pprint.pprint(sampleDrop)
val = dead.pickup(sampleDrop['key'])
pprint.pprint(val)
mock_pymongo.dead.drop.find_one_and_delete.assert_called_with({"key": sampleDrop['key']})
assert sampleDrop["data"] == val
@patch('pymongo.MongoClient')
@freeze_time("2012-01-14")
def test_track_updated_when_accessed(mock_pymongo):
sampleDrop = get_sample_drop()
mock_pymongo.dead.drop.find_one_and_delete.return_value = sampleDrop
dead = DropHandler(mock_pymongo)
val = dead.pickup(sampleDrop['key'])
mock_pymongo.dead.track.update.assert_called_with({"key": sampleDrop['key']}, {"$set":{"pickedUp":datetime.datetime(2012, 1, 14)},"$unset":{"key":""}})
assert sampleDrop["data"] == val
def get_sample_stats():
samplestats =[
{ "_id" : { "year" : 2018, "month" : 3, "day" : 2 }, "count" : 67, "distinctCount" : 49 },
{ "_id" : { "year" : 2018, "month" : 3, "day" : 3 }, "count" : 24, "distinctCount" : 15 }
]
return samplestats
@patch('pymongo.MongoClient')
def test_stats_returned(mock_pymongo):
#db.track.aggregate({ $group: { _id: { "year":{$year:"$createdDate"},"month":{$month:"$createdDate"},"day":{$dayOfMonth: "$createdDate"}},createdDate: { $sum: 1 } } })
mock_pymongo.dead.track.find.return_value= get_sample_stats()
dead = DropHandler(mock_pymongo)
val = dead.stats()
expected =[
{"$group":
{ "_id": { "year":{"$year":"$createdDate"},"month":{"$month":"$createdDate"},"day":{"$dayOfMonth": "$createdDate"},"userHash":"$userHash"},
"count": { "$sum": 1 } ,
}
},
{"$group":
{ "_id": { "year":"$_id.year","month":"$_id.month","day":"$_id.day"},
"count": { "$sum": "$count" },
"distinctCount": { "$sum": 1 }
}
},
{"$sort": {"_id.year":1,"_id.month":1,"_id.day":1}},
];
mock_pymongo.dead.track.aggregate.assert_called_with(expected)
@patch('pymongo.MongoClient')
@freeze_time("2012-01-14")
def test_drop_not_retruned_when_no_create_date(mock_pymongo):
# to handle old drops
sampleDrop = get_sample_drop()
sampleDrop.pop('createdDate')
mock_pymongo.dead.drop.find_one_and_delete.return_value=sampleDrop
dead = DropHandler(mock_pymongo)
val = dead.pickup(sampleDrop['key'])
pprint.pprint(sampleDrop)
pprint.pprint(val)
mock_pymongo.dead.drop.find_one_and_delete.assert_called_with({"key": sampleDrop['key']})
assert val == []
def get_sample_drop():
return {'key':'12345','data':"test data return","createdDate":datetime.datetime.now()}
@patch('pymongo.MongoClient')
@freeze_time("2012-01-14")
def test_drop_deleted_and_not_returned_when_24hours_old(mock_pymongo):
key = "anything"
sampleDrop = get_sample_drop()
sampleDrop["createdDate"] = datetime.datetime(2012, 1, 12)
mock_pymongo.dead.drop.find.return_value=[sampleDrop]
dead = DropHandler(mock_pymongo)
val = dead.pickup(key)
assert val == []
mock_pymongo.dead.drop.find_one_and_delete.assert_called_with({"key": key})
@patch('pymongo.MongoClient')
def test_return_none_when_not_existing(mock_pymongo):
sampleDrop = get_sample_drop()
mock_pymongo.dead.drop.find_one_and_delete.return_value=[]
dead = DropHandler(mock_pymongo)
val = dead.pickup(sampleDrop['key'])
assert val == []
mock_pymongo.dead.drop.find_one_and_delete.assert_called_with({"key": sampleDrop['key']})
@patch('pymongo.MongoClient')
def test_timed_key_is_saved(mock_pymongo):
dead = DropHandler(mock_pymongo)
timed_key = dead.get_timed_key()
mock_pymongo.dead.formKeys.insert_one.assert_called_with({"key": timed_key,"created": ANY})