-
Notifications
You must be signed in to change notification settings - Fork 5
Expand file tree
/
Copy pathtest_cdx.py
More file actions
117 lines (93 loc) · 3.78 KB
/
Copy pathtest_cdx.py
File metadata and controls
117 lines (93 loc) · 3.78 KB
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
import os
from iterable.datatypes import CDXIterable
class TestCDX:
def test_id(self):
datatype_id = CDXIterable.id()
assert datatype_id == "cdx"
def test_flatonly(self):
flag = CDXIterable.is_flatonly()
assert flag
def test_has_totals(self):
flag = CDXIterable.has_totals()
assert flag
def test_openclose(self):
"""Test basic open/close"""
test_file = "testdata/test_cdx.cdx"
os.makedirs("testdata", exist_ok=True)
with open(test_file, "w") as f:
f.write("http://example.com 20240101120000 http://example.com text/html 200 -\n")
iterable = CDXIterable(test_file)
iterable.close()
if os.path.exists(test_file):
os.unlink(test_file)
def test_read_one(self):
"""Test reading single CDX record"""
test_file = "testdata/test_cdx_read.cdx"
os.makedirs("testdata", exist_ok=True)
with open(test_file, "w") as f:
f.write("http://example.com 20240101120000 http://example.com text/html 200 - - file.warc 0\n")
iterable = CDXIterable(test_file)
record = iterable.read()
assert isinstance(record, dict)
assert "url" in record or "timestamp" in record
iterable.close()
if os.path.exists(test_file):
os.unlink(test_file)
def test_read_bulk(self):
"""Test reading bulk CDX records"""
test_file = "testdata/test_cdx_bulk.cdx"
os.makedirs("testdata", exist_ok=True)
with open(test_file, "w") as f:
f.write("http://example.com 20240101120000 http://example.com text/html 200 - - file.warc 0\n")
f.write("http://example.org 20240101120001 http://example.org text/html 200 - - file.warc 100\n")
iterable = CDXIterable(test_file)
chunk = iterable.read_bulk(2)
assert isinstance(chunk, list)
assert len(chunk) == 2
iterable.close()
if os.path.exists(test_file):
os.unlink(test_file)
def test_totals(self):
"""Test totals() method"""
test_file = "testdata/test_cdx_totals.cdx"
os.makedirs("testdata", exist_ok=True)
with open(test_file, "w") as f:
f.write("http://example.com 20240101120000 http://example.com text/html 200 - - file.warc 0\n")
f.write("http://example.org 20240101120001 http://example.org text/html 200 - - file.warc 100\n")
iterable = CDXIterable(test_file)
total = iterable.totals()
assert total >= 2
iterable.close()
if os.path.exists(test_file):
os.unlink(test_file)
def test_reset(self):
"""Test reset functionality"""
test_file = "testdata/test_cdx_reset.cdx"
os.makedirs("testdata", exist_ok=True)
with open(test_file, "w") as f:
f.write("http://example.com 20240101120000 http://example.com text/html 200 - - file.warc 0\n")
iterable = CDXIterable(test_file)
record1 = iterable.read()
iterable.reset()
record2 = iterable.read()
assert record1 == record2
iterable.close()
if os.path.exists(test_file):
os.unlink(test_file)
def test_write(self):
"""Test writing CDX record"""
test_file = "testdata/test_cdx_write.cdx"
os.makedirs("testdata", exist_ok=True)
iterable = CDXIterable(test_file, mode="w")
record = {
"url": "http://example.com",
"timestamp": "20240101120000",
"original_url": "http://example.com",
"mime_type": "text/html",
"status_code": "200",
}
iterable.write(record)
iterable.close()
assert os.path.exists(test_file)
if os.path.exists(test_file):
os.unlink(test_file)