2017-10-19 22:11:22 +00:00
|
|
|
import mock
|
2017-10-13 17:44:07 +00:00
|
|
|
from warcprox.dedup import CdxServerDedup
|
|
|
|
|
|
|
|
|
2017-10-19 22:11:22 +00:00
|
|
|
def test_cdx_dedup():
|
|
|
|
# Mock CDX Server responses to simulate found, not found and errors.
|
2018-01-15 20:58:36 +00:00
|
|
|
url = "http://example.com"
|
|
|
|
# not found case
|
|
|
|
result = mock.Mock()
|
|
|
|
result.status = 200
|
|
|
|
result.data = b'20170101020405 test'
|
|
|
|
cdx_server = CdxServerDedup(cdx_url="dummy-cdx-server-url")
|
|
|
|
cdx_server.http_pool.request = mock.MagicMock(return_value=result)
|
|
|
|
res = cdx_server.lookup(digest_key="B2LTWWPUOYAH7UIPQ7ZUPQ4VMBSVC36A",
|
|
|
|
url=url)
|
|
|
|
assert res is None
|
2017-10-19 22:11:22 +00:00
|
|
|
|
2018-01-15 20:58:36 +00:00
|
|
|
# found case
|
|
|
|
result = mock.Mock()
|
|
|
|
result.status = 200
|
|
|
|
result.data = b'20170203040503 B2LTWWPUOYAH7UIPQ7ZUPQ4VMBSVC36A'
|
|
|
|
cdx_server = CdxServerDedup(cdx_url="dummy-cdx-server-url")
|
|
|
|
cdx_server.http_pool.request = mock.MagicMock(return_value=result)
|
|
|
|
res = cdx_server.lookup(digest_key="B2LTWWPUOYAH7UIPQ7ZUPQ4VMBSVC36A",
|
|
|
|
url=url)
|
|
|
|
assert res["date"] == b"2017-02-03T04:05:03Z"
|
2017-10-19 22:11:22 +00:00
|
|
|
|
2018-01-15 20:58:36 +00:00
|
|
|
# invalid CDX result status code
|
|
|
|
result = mock.Mock()
|
|
|
|
result.status = 400
|
|
|
|
result.data = b'20170101020405 B2LTWWPUOYAH7UIPQ7ZUPQ4VMBSVC36A'
|
|
|
|
cdx_server = CdxServerDedup(cdx_url="dummy-cdx-server-url")
|
|
|
|
cdx_server.http_pool.request = mock.MagicMock(return_value=result)
|
|
|
|
res = cdx_server.lookup(digest_key="B2LTWWPUOYAH7UIPQ7ZUPQ4VMBSVC36A",
|
|
|
|
url=url)
|
|
|
|
assert res is None
|
|
|
|
|
|
|
|
# invalid CDX result content
|
|
|
|
result = mock.Mock()
|
|
|
|
result.status = 200
|
|
|
|
result.data = b'InvalidExceptionResult'
|
|
|
|
cdx_server = CdxServerDedup(cdx_url="dummy-cdx-server-url")
|
|
|
|
cdx_server.http_pool.request = mock.MagicMock(return_value=result)
|
|
|
|
res = cdx_server.lookup(digest_key="B2LTWWPUOYAH7UIPQ7ZUPQ4VMBSVC36A",
|
|
|
|
url=url)
|
|
|
|
assert res is None
|