Changeset View
Changeset View
Standalone View
Standalone View
swh/provenance/tests/mongo/test_backend.py
- This file was added.
# Copyright (C) 2021 The Software Heritage developers | |||||
# See the AUTHORS file at the top-level directory of this distribution | |||||
# License: GNU General Public License version 3, or any later version | |||||
# See top-level LICENSE file for more information | |||||
""" | |||||
Unit tests for the mongo backend | |||||
""" | |||||
from datetime import datetime | |||||
import pymongo.database | |||||
import pytest | |||||
# from swh.provenance.mongo.backend import ProvenanceStorageMongoDb | |||||
from swh.provenance import get_provenance_storage | |||||
from swh.provenance.interface import RevisionData | |||||
class TestMongoDBInit: | |||||
""" | |||||
Make sure mongo indexes are set | |||||
Validate the datamodel | |||||
mongomock is used to simulate the mongo server. | |||||
Version assumed is mongo 5.0 engine:wiredTiger | |||||
""" | |||||
def test_contnet_sha1_unique_index(self, mongodb): | |||||
pass | |||||
def test_contnet_sha1_ts_combination_index(self, mongodb): | |||||
pass | |||||
def test_directory_sha1_unique_index(self, mongodb): | |||||
pass | |||||
def test_directory_sha1_ts_combination_index(self, mongodb): | |||||
pass | |||||
def test_origin_sha1_unique_index(self, mongodb): | |||||
pass | |||||
class TestMongoBackend: | |||||
""" | |||||
Test mongo backend | |||||
This class tests each method in mongodb backend in isolation | |||||
methods are verified by directly interacting with the mongoDB | |||||
mongomock is used to simulate the mongo server. | |||||
""" | |||||
# FIXME, many, if not all, of the following methods can be refactored | |||||
# to make this test class backend agnostic | |||||
# FIMXE, use fixtures to supply test data (along with the performance tests) | |||||
# FIXME, consider splitting to different classes or modules if this gets too long | |||||
@pytest.fixture | |||||
def backend(self, mongodb: pymongo.database.Database): | |||||
mongodb_params = { | |||||
"dbname": mongodb.name, | |||||
} | |||||
from pymongo import MongoClient | |||||
mongo_client = MongoClient("mongodb://localhost:27017") | |||||
mongo_client.drop_database(mongodb.name) | |||||
with get_provenance_storage( | |||||
cls="mongodb", db=mongodb_params, engine="mongomock" | |||||
) as storage: | |||||
self._set_indexes(storage) | |||||
yield storage | |||||
def _set_indexes(self, backend): | |||||
backend.create_indexes("content", [("sha1", 1)], unique=True) | |||||
backend.create_indexes("content", [("sha1", 1), ("ts", 1)]) | |||||
backend.create_indexes("directory", [("sha1", 1)], unique=True) | |||||
backend.create_indexes("directory", [("sha1", 1), ("ts", 1)]) | |||||
backend.create_indexes("origin", [("sha1", 1)], unique=True) | |||||
backend.create_indexes("revision", [("sha1", 1)], unique=True) | |||||
# add content tests | |||||
def test_add_content_empty(self, backend): | |||||
assert backend.content_add({}) is True | |||||
results = backend.entity_get("content") | |||||
assert len(results) == 0 | |||||
assert backend.content_add([]) is True | |||||
results = backend.entity_get("content") | |||||
assert len(results) == 0 | |||||
def test_add_content_with_insert(self, backend): | |||||
# add data using add_contnet | |||||
# get data from mongo and compare | |||||
sha1_1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
sha1_2 = "a94a8fe5ccb19ba61c4c0873d391e987982fbbd3" | |||||
data = { | |||||
sha1_1: None, | |||||
sha1_2: datetime(2020, 10, 10), | |||||
} # two new rcds will be inserted | |||||
assert backend.content_add(data) is True | |||||
results = backend.entity_get("content") | |||||
assert len(results) == 2 | |||||
results = backend.entity_get("content", {"sha1": sha1_1}) | |||||
assert len(results) == 1 | |||||
cnt = results[0] | |||||
assert cnt["ts"] is None | |||||
assert cnt["revision"] == {} | |||||
assert cnt["directory"] == {} | |||||
results = backend.entity_get("content", {"sha1": sha1_2}) | |||||
assert len(results) == 1 | |||||
cnt = results[0] | |||||
assert cnt["ts"] == datetime(2020, 10, 10).timestamp() | |||||
assert cnt["revision"] == {} | |||||
assert cnt["directory"] == {} | |||||
def test_add_content_with_update_later_date(self, backend): | |||||
sha1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
revision = {"test": "test"} | |||||
backend.entity_add( | |||||
"content", | |||||
[ | |||||
{ | |||||
"sha1": sha1, | |||||
"ts": datetime(2020, 10, 10).timestamp(), | |||||
"revision": revision, | |||||
"directory": {}, | |||||
} | |||||
], | |||||
) | |||||
new_date = datetime(2010, 10, 10) | |||||
data = {sha1: new_date} | |||||
# data has a date earlier than the one in the db, so the date will be updated | |||||
assert backend.content_add(data) is True | |||||
results = backend.entity_get("content") | |||||
assert len(results) == 1 | |||||
results = backend.entity_get("content", {"sha1": sha1}) | |||||
assert len(results) == 1 | |||||
cnt = results[0] | |||||
assert cnt["ts"] == new_date.timestamp() | |||||
assert cnt["revision"] == revision | |||||
assert cnt["directory"] == {} | |||||
def test_add_content_with_update_none_date(self, backend): | |||||
sha1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
revision = {"test": "test"} | |||||
backend.entity_add( | |||||
"content", | |||||
[{"sha1": sha1, "ts": None, "revision": revision, "directory": {}}], | |||||
) | |||||
new_date = datetime(2010, 10, 10) | |||||
data = {sha1: new_date} | |||||
# db has None date, will be updated with the newly supplied date | |||||
assert backend.content_add(data) is True | |||||
results = backend.entity_get("content", {"sha1": sha1}) | |||||
assert len(results) == 1 | |||||
cnt = results[0] | |||||
assert cnt["ts"] == new_date.timestamp() | |||||
assert cnt["revision"] == revision | |||||
assert cnt["directory"] == {} | |||||
def test_add_content_do_not_update_older_date(self, backend): | |||||
sha1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
revision = {"test": "test"} | |||||
backend.entity_add( | |||||
"content", | |||||
[ | |||||
{ | |||||
"sha1": sha1, | |||||
"ts": datetime(2010, 10, 10).timestamp(), | |||||
"revision": revision, | |||||
"directory": {}, | |||||
} | |||||
], | |||||
) | |||||
new_date = datetime(2020, 10, 10) | |||||
data = {sha1: new_date} | |||||
# data has a date later than the one in the db, no update will happen | |||||
assert backend.content_add(data) is True | |||||
results = backend.entity_get("content", {"sha1": sha1}) | |||||
assert len(results) == 1 | |||||
cnt = results[0] | |||||
assert cnt["ts"] == datetime(2010, 10, 10).timestamp() | |||||
def test_add_content_multiple(self, backend): | |||||
sha1_1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
sha1_2 = "a94a8fe5ccb19ba61c4c0873d391e987982fbbd3" | |||||
sha1_3 = "109f4b3c50d7b0df729d299bc6f8e9ef9066971f" | |||||
sha1_4 = "3ebfa301dc59196f18593c45e519287a23297589" | |||||
revision = {"test": "test"} | |||||
backend.entity_add( | |||||
"content", | |||||
[ | |||||
{ | |||||
"sha1": sha1_1, | |||||
"ts": 1286661600, | |||||
"revision": revision, | |||||
"directory": {}, | |||||
}, | |||||
{"sha1": sha1_2, "ts": None, "revision": revision, "directory": {}}, | |||||
{ | |||||
"sha1": sha1_3, | |||||
"ts": 1631889655, | |||||
"revision": revision, | |||||
"directory": {}, | |||||
}, | |||||
], | |||||
) | |||||
data = { | |||||
sha1_1: datetime(2020, 10, 10), # given date is in the future, no update | |||||
sha1_2: datetime(2020, 10, 10), # will update None date | |||||
sha1_3: datetime(2010, 10, 10), # date in the past, will update | |||||
sha1_4: datetime(2020, 10, 10), # new rcd, will insert | |||||
} | |||||
assert backend.content_add(data) is True | |||||
results = backend.entity_get("content") | |||||
assert len(results) == 4 | |||||
cnt = backend.entity_get("content", {"sha1": sha1_1})[0] | |||||
assert cnt["ts"] == 1286661600 | |||||
assert cnt["revision"] == revision | |||||
assert cnt["directory"] == {} | |||||
cnt = backend.entity_get("content", {"sha1": sha1_2})[0] | |||||
assert cnt["ts"] == datetime(2020, 10, 10).timestamp() | |||||
assert cnt["revision"] == revision | |||||
assert cnt["directory"] == {} | |||||
cnt = backend.entity_get("content", {"sha1": sha1_3})[0] | |||||
assert cnt["ts"] == datetime(2010, 10, 10).timestamp() | |||||
assert cnt["revision"] == revision | |||||
assert cnt["directory"] == {} | |||||
cnt = backend.entity_get("content", {"sha1": sha1_4})[0] | |||||
assert cnt["ts"] == datetime(2020, 10, 10).timestamp() | |||||
assert cnt["revision"] == {} | |||||
assert cnt["directory"] == {} | |||||
# # # add directory tests | |||||
def test_add_directory_empty(self, backend): | |||||
assert backend.directory_add({}) is True | |||||
results = backend.entity_get("directory") | |||||
assert len(results) == 0 | |||||
assert backend.directory_add([]) is True | |||||
results = backend.entity_get("directory") | |||||
assert len(results) == 0 | |||||
def test_add_directory_with_insert(self, backend): | |||||
# add data using add_directory | |||||
# get data from mongo and compare | |||||
sha1_1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
sha1_2 = "a94a8fe5ccb19ba61c4c0873d391e987982fbbd3" | |||||
data = { | |||||
sha1_1: None, | |||||
sha1_2: datetime(2020, 10, 10), | |||||
} # two new rcds will be inserted | |||||
assert backend.directory_add(data) is True | |||||
results = backend.entity_get("directory") | |||||
assert len(results) == 2 | |||||
cnt = backend.entity_get("directory", {"sha1": sha1_1})[0] | |||||
assert cnt["ts"] is None | |||||
assert cnt["revision"] == {} | |||||
cnt = backend.entity_get("directory", {"sha1": sha1_2})[0] | |||||
assert cnt["ts"] == datetime(2020, 10, 10).timestamp() | |||||
assert cnt["revision"] == {} | |||||
def test_add_directory_with_update_later_date(self, backend): | |||||
sha1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
revision = {"test": "test"} | |||||
backend.entity_add( | |||||
"directory", [{"sha1": sha1, "ts": 1631881748, "revision": revision}] | |||||
) | |||||
new_date = datetime(2010, 10, 10) | |||||
data = {sha1: new_date} | |||||
# data has a date earlier than the one in the db, will update the date | |||||
assert backend.directory_add(data) is True | |||||
diy = backend.entity_get("directory", {"sha1": sha1})[0] | |||||
assert diy["ts"] == new_date.timestamp() | |||||
assert diy["revision"] == revision | |||||
def test_add_directory_with_update_none_date(self, backend): | |||||
sha1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
revision = {"test": "test"} | |||||
backend.entity_add( | |||||
"directory", [{"sha1": sha1, "ts": None, "revision": revision}] | |||||
) | |||||
new_date = datetime(2010, 10, 10) | |||||
data = {sha1: new_date} | |||||
# db has None date, will be updated with the given date | |||||
assert backend.directory_add(data) is True | |||||
diy = backend.entity_get("directory", {"sha1": sha1})[0] | |||||
assert diy["ts"] == new_date.timestamp() | |||||
assert diy["revision"] == revision | |||||
def test_add_directory_do_not_update_older_date(self, backend): | |||||
sha1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
revision = {"test": "test"} | |||||
backend.entity_add( | |||||
"directory", | |||||
[ | |||||
{ | |||||
"sha1": sha1, | |||||
"ts": datetime(2010, 10, 10).timestamp(), | |||||
"revision": revision, | |||||
} | |||||
], | |||||
) | |||||
new_date = datetime(2020, 10, 10) | |||||
data = {sha1: new_date} | |||||
# data has a date later than the one in the db, no update will happen | |||||
assert backend.directory_add(data) is True | |||||
results = backend.entity_get("directory", {"sha1": sha1}) | |||||
assert len(results) == 1 | |||||
cnt = results[0] | |||||
assert cnt["ts"] == datetime(2010, 10, 10).timestamp() | |||||
def test_add_directory_multiple(self, backend): | |||||
sha1_1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
sha1_2 = "a94a8fe5ccb19ba61c4c0873d391e987982fbbd3" | |||||
sha1_3 = "109f4b3c50d7b0df729d299bc6f8e9ef9066971f" | |||||
sha1_4 = "3ebfa301dc59196f18593c45e519287a23297589" | |||||
revision = {"test": "test"} | |||||
backend.entity_add( | |||||
"directory", | |||||
[ | |||||
{"sha1": sha1_1, "ts": 1286661600, "revision": revision}, | |||||
{"sha1": sha1_2, "ts": None, "revision": revision}, | |||||
{"sha1": sha1_3, "ts": 1631889655, "revision": revision}, | |||||
], | |||||
) | |||||
data = { | |||||
sha1_1: datetime(2020, 10, 10), # given date is in the future, no update | |||||
sha1_2: datetime(2020, 10, 10), # will update None date | |||||
sha1_3: datetime(2010, 10, 10), # date in the past, will update | |||||
sha1_4: datetime(2010, 10, 10), # new rcd, will insert | |||||
} | |||||
assert backend.directory_add(data) is True | |||||
results = backend.entity_get("directory") | |||||
assert len(results) == 4 | |||||
dry = backend.entity_get("directory", {"sha1": sha1_1})[0] | |||||
assert dry["ts"] == 1286661600 | |||||
assert dry["revision"] == revision | |||||
dry = backend.entity_get("directory", {"sha1": sha1_2})[0] | |||||
assert dry["ts"] == datetime(2020, 10, 10).timestamp() | |||||
assert dry["revision"] == revision | |||||
dry = backend.entity_get("directory", {"sha1": sha1_3})[0] | |||||
assert dry["ts"] == datetime(2010, 10, 10).timestamp() | |||||
assert dry["revision"] == revision | |||||
dry = backend.entity_get("directory", {"sha1": sha1_4})[0] | |||||
assert dry["ts"] == datetime(2010, 10, 10).timestamp() | |||||
assert dry["revision"] == {} | |||||
# # # content_get tests | |||||
def test_content_get_empty_ids(self, backend): | |||||
sha1_1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
sha1_2 = "a94a8fe5ccb19ba61c4c0873d391e987982fbbd3" | |||||
data1 = {"sha1": sha1_1, "ts": datetime(2010, 10, 8).timestamp()} | |||||
data2 = {"sha1": sha1_2, "ts": datetime(2020, 8, 20).timestamp()} | |||||
backend.entity_add("content", [data1, data2]) | |||||
results = backend.content_get([]) | |||||
assert results == {} | |||||
def test_content_get(self, backend): | |||||
sha1_1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
sha1_2 = "a94a8fe5ccb19ba61c4c0873d391e987982fbbd3" | |||||
sha1_3 = "109f4b3c50d7b0df729d299bc6f8e9ef9066971f" | |||||
data1 = {"sha1": sha1_1, "ts": datetime(2010, 10, 8).timestamp()} | |||||
data2 = {"sha1": sha1_2, "ts": datetime(2020, 8, 20).timestamp()} | |||||
# This has None for ts, will not be returend from content_get | |||||
data3 = {"sha1": sha1_2, "ts": None} | |||||
backend.entity_add("content", [data1, data2, data3]) | |||||
results = backend.content_get([sha1_1, sha1_2, sha1_3]) | |||||
assert len(results) == 2 | |||||
results[sha1_1] = datetime(2010, 10, 8).timestamp() | |||||
results[sha1_2] = datetime(2020, 8, 20).timestamp() | |||||
# # # directory_get tests | |||||
def test_directory_get_empty_ids(self, backend): | |||||
sha1_1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
sha1_2 = "a94a8fe5ccb19ba61c4c0873d391e987982fbbd3" | |||||
data1 = {"sha1": sha1_1, "ts": datetime(2010, 10, 8).timestamp()} | |||||
data2 = {"sha1": sha1_2, "ts": datetime(2020, 8, 20).timestamp()} | |||||
backend.entity_add("directory", [data1, data2]) | |||||
results = backend.directory_get([]) | |||||
assert results == {} | |||||
def test_directory_get(self, backend): | |||||
sha1_1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
sha1_2 = "a94a8fe5ccb19ba61c4c0873d391e987982fbbd3" | |||||
sha1_3 = "109f4b3c50d7b0df729d299bc6f8e9ef9066971f" | |||||
data1 = {"sha1": sha1_1, "ts": datetime(2010, 10, 8).timestamp()} | |||||
data2 = {"sha1": sha1_2, "ts": datetime(2020, 8, 20).timestamp()} | |||||
# This has None for ts, will not be returend from directory_get | |||||
data3 = {"sha1": sha1_2, "ts": None} | |||||
backend.entity_add("directory", [data1, data2, data3]) | |||||
results = backend.directory_get([sha1_1, sha1_2, sha1_3]) | |||||
assert len(results) == 2 | |||||
results[sha1_1] = datetime(2010, 10, 8).timestamp() | |||||
results[sha1_2] = datetime(2020, 8, 20).timestamp() | |||||
# # # location_add tests | |||||
def test_location_add(self, backend): | |||||
# FIXME, this will change with model change | |||||
assert backend.location_add([]) is True | |||||
# # origin_add tests | |||||
def test_origin_add_empty(self, backend): | |||||
assert backend.origin_add({}) is True | |||||
results = backend.entity_get("origin") | |||||
assert len(results) == 0 | |||||
def test_origin_add_new(self, backend): | |||||
sha1_1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
sha1_2 = "a94a8fe5ccb19ba61c4c0873d391e987982fbbd3" | |||||
data = {sha1_1: "1.example.com", sha1_2: "2.example.com"} | |||||
assert backend.origin_add(data) is True | |||||
results = backend.entity_get("origin") | |||||
assert len(results) == 2 | |||||
def test_origin_add_skip_existing(self, backend): | |||||
# sending an existing origin hash will not add or update any record | |||||
sha1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
backend.entity_add("origin", [{"sha1": sha1, "url": "example.com"}]) | |||||
sha1_1 = "a94a8fe5ccb19ba61c4c0873d391e987982fbbd3" | |||||
data = {sha1: "1.example.com", sha1_1: "2.example.com"} | |||||
assert backend.origin_add(data) is True | |||||
results = backend.entity_get("origin") | |||||
assert len(results) == 2 | |||||
origin = backend.entity_get("origin", {"sha1": sha1})[0] | |||||
assert origin["url"] == "example.com" # not 1.example.com | |||||
# # # origin_get tests | |||||
def test_origin_get_empty(self, backend): | |||||
sha1_1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
sha1_2 = "a94a8fe5ccb19ba61c4c0873d391e987982fbbd3" | |||||
backend.entity_add( | |||||
"origin", | |||||
[ | |||||
{"sha1": sha1_1, "url": "1.example.com"}, | |||||
{"sha1": sha1_2, "url": "2.example.com"}, | |||||
], | |||||
) | |||||
assert backend.origin_get({}) == {} | |||||
def test_origin_get(self, backend): | |||||
sha1_1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
sha1_2 = "a94a8fe5ccb19ba61c4c0873d391e987982fbbd3" | |||||
backend.entity_add( | |||||
"origin", | |||||
[ | |||||
{"sha1": sha1_1, "url": "1.example.com"}, | |||||
{"sha1": sha1_2, "url": "2.example.com"}, | |||||
], | |||||
) | |||||
assert backend.origin_get([sha1_1]) == {sha1_1: "1.example.com"} | |||||
# # # revision_add tests | |||||
def test_revision_add_empty(self, backend): | |||||
assert backend.revision_add({}) is True | |||||
assert backend.revision_add([]) is True | |||||
def test_revision_add_create_revision_with_empty_date(self, backend): | |||||
""" | |||||
Input to revision_add is a list of sha1s | |||||
RevisionData will be considerd as None | |||||
""" | |||||
sha1_1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
sha1_2 = "a94a8fe5ccb19ba61c4c0873d391e987982fbbd3" | |||||
data = [sha1_1, sha1_2] | |||||
assert backend.revision_add(data) is True | |||||
results = backend.entity_get("revision") | |||||
assert len(results) == 2 | |||||
revision_1 = backend.entity_get("revision", {"sha1": sha1_1})[0] | |||||
assert revision_1["preferred"] is None | |||||
assert revision_1["origin"] == [] | |||||
assert revision_1["revision"] == [] | |||||
assert revision_1["ts"] is None | |||||
def test_revision_add_create_new_revision_with_date(self, backend): | |||||
""" | |||||
RevisionData is a dataclass with date and sha1 of the preferred origin | |||||
""" | |||||
sha1_1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
sha1_2 = "a94a8fe5ccb19ba61c4c0873d391e987982fbbd3" | |||||
origin_sha1 = "109f4b3c50d7b0df729d299bc6f8e9ef9066971f" | |||||
# DOUBT, this function is not expecting preferred origin to be already added | |||||
# what if an unknown origin comes as preferred, an integrity error? | |||||
rev_data_1 = RevisionData(datetime(2020, 8, 20), origin_sha1) | |||||
rev_data_2 = RevisionData(datetime(2010, 8, 20), None) | |||||
data = {sha1_1: rev_data_1, sha1_2: rev_data_2} | |||||
assert backend.revision_add(data) is True | |||||
results = backend.entity_get("revision") | |||||
assert len(results) == 2 | |||||
revision_1 = backend.entity_get("revision", {"sha1": sha1_1})[0] | |||||
assert revision_1["preferred"] == rev_data_1.origin | |||||
assert ( | |||||
revision_1["origin"] == [] | |||||
) # DOUBT, should this contain the new preferred origin? | |||||
assert ( | |||||
revision_1["revision"] == [] | |||||
) # DOUBT, this is the history, should the new revaion should be added here? | |||||
assert revision_1["ts"] == rev_data_1.date.timestamp() | |||||
revision_2 = backend.entity_get("revision", {"sha1": sha1_2})[0] | |||||
assert revision_2["preferred"] is None | |||||
assert revision_2["origin"] == [] | |||||
assert revision_2["revision"] == [] | |||||
assert revision_2["ts"] == datetime(2010, 8, 20).timestamp() | |||||
def test_revision_add_update_none_date_and_origin(self, backend): | |||||
sha1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
# add a revision with sha1 | |||||
backend.entity_add( | |||||
"revision", | |||||
[ | |||||
{ | |||||
"sha1": sha1, | |||||
"ts": None, | |||||
"preferred": None, | |||||
"origin": ["test-origin"], # FIXME, validate sha1 at the datamodel | |||||
"revision": [ | |||||
"test-revision" | |||||
], # FIXME, validate sha1 at the datamodel | |||||
} | |||||
], | |||||
) | |||||
origin_sha1 = "109f4b3c50d7b0df729d299bc6f8e9ef9066971f" | |||||
rev_data = RevisionData(datetime(2010, 8, 20), origin_sha1) | |||||
data = {sha1: rev_data} | |||||
# both date and origin will be updated since db has None for both | |||||
assert backend.revision_add(data) is True | |||||
# making sure an update happend and no duplicate record is added | |||||
assert len(backend.entity_get("revision", {"sha1": sha1})) == 1 | |||||
revision = backend.entity_get("revision", {"sha1": sha1})[0] | |||||
assert revision["ts"] == datetime(2010, 8, 20).timestamp() | |||||
assert revision["preferred"] == origin_sha1 | |||||
assert revision["origin"] == ["test-origin"] | |||||
assert revision["revision"] == ["test-revision"] | |||||
def test_revision_add_existing_sha1_with_earlier_date(self, backend): | |||||
# date will be updated with the new (earlier) date | |||||
sha1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
# add a revision with sha1 | |||||
backend.entity_add( | |||||
"revision", | |||||
[ | |||||
{ | |||||
"sha1": sha1, | |||||
"ts": datetime(2020, 8, 20).timestamp(), | |||||
"preferred": None, | |||||
"origin": ["test-origin"], # FIXME, validate sha1 at the datamodel | |||||
"revision": [ | |||||
"test-revision" | |||||
], # FIXME, validate sha1 at the datamodel | |||||
} | |||||
], | |||||
) | |||||
# date in data is earlier then the one in db | |||||
rev_data = RevisionData(datetime(2010, 8, 20), None) | |||||
data = {sha1: rev_data} | |||||
assert backend.revision_add(data) is True | |||||
# making sure an update happend and no duplicate record is added | |||||
results = backend.entity_get("revision") | |||||
assert len(results) == 1 | |||||
revision = backend.entity_get("revision", {"sha1": sha1})[0] | |||||
assert revision["ts"] == datetime(2010, 8, 20).timestamp() | |||||
assert revision["preferred"] is None | |||||
def test_revision_add_existing_sha1_with_later_date(self, backend): | |||||
# date will not be updated as it is later than the one in the db | |||||
# preferred revision will be updated | |||||
sha1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
# add a revision with sha1 | |||||
backend.entity_add( | |||||
"revision", | |||||
[ | |||||
{ | |||||
"sha1": sha1, | |||||
"ts": datetime(2010, 8, 20).timestamp(), | |||||
"preferred": None, | |||||
"origin": ["test-origin"], # FIXME, validate sha1 at the datamodel | |||||
"revision": [ | |||||
"test-revision" | |||||
], # FIXME, validate sha1 at the datamodel | |||||
} | |||||
], | |||||
) | |||||
# date in data is earlier then the one in db | |||||
origin_sha1 = "109f4b3c50d7b0df729d299bc6f8e9ef9066971f" | |||||
rev_data = RevisionData(datetime(2020, 8, 20), origin_sha1) | |||||
data = {sha1: rev_data} | |||||
assert backend.revision_add(data) is True | |||||
# making sure an update happend and no duplicate record is added | |||||
assert len(backend.entity_get("revision", {"sha1": sha1})) == 1 | |||||
revision = backend.entity_get("revision", {"sha1": sha1})[0] | |||||
assert revision["ts"] == datetime(2010, 8, 20).timestamp() | |||||
assert revision["preferred"] == origin_sha1 | |||||
def test_revision_add_multiple(self, backend): | |||||
# Both ts and origin will be updeated | |||||
sha1_1 = "cf23df2207d99a74fbe169e3eba035e633b65d94" | |||||
sha1_2 = "" | |||||
sha1_3 = "" | |||||
backend.entity_add( | |||||
"revision", | |||||
[ | |||||
{ | |||||
"sha1": sha1_1, | |||||
"ts": None, | |||||
"preferred": None, | |||||
"origin": ["test-origin"], | |||||
"revision": ["test-revision"], | |||||
}, | |||||
{ # only origin will be updeated | |||||
"sha1": sha1_2, | |||||
"ts": datetime(2010, 8, 20).timestamp(), | |||||
"preferred": None, | |||||
"origin": ["test-origin"], | |||||
"revision": ["test-revision"], | |||||
}, | |||||
{ # only date will be updated | |||||
"sha1": sha1_3, | |||||
"ts": None, | |||||
"preferred": "109f4b3c50d7b0df729d299bc6f8e9ef9066971f", | |||||
"origin": ["test-origin"], | |||||
"revision": ["test-revision"], | |||||
}, | |||||
], | |||||
) | |||||
# both date and origin will be updated | |||||
# rev_data_1 = RevisionData( | |||||
# datetime(2020, 8, 20), "109f4b3c50d7b0df729d299bc6f8e9ef9066971f" | |||||
# ) | |||||
# # only origin will be updated | |||||
# rev_data_2 = RevisionData(None, "109f4b3c50d7b0df729d299bc6f8e9ef9066971f") | |||||
# # only date will be updated | |||||
# rev_data_3 = RevisionData( | |||||
# datetime(2020, 8, 20), "109f4b3c50d7b0df729d299bc6f8e9ef9066971f" | |||||
# ) | |||||
# # this record will be inserted | |||||
# rev_data_4 = RevisionData( | |||||
# datetime(2020, 8, 20), "109f4b3c50d7b0df729d299bc6f8e9ef9066971f" | |||||
# ) | |||||
# data = {sha1_1: rev_data_1, sha1_2: rev_data_2, sha1_3: rev_data_3, | |||||
# sha1_4: rev_data_4} | |||||
# FiXME, compare | |||||
# # # revision_get tests | |||||
def test_revision_get(self, backend): | |||||
# create data with preferred as None, ts None | |||||
# preferred as Not None, ts not None | |||||
pass | |||||
def test_revision_get_preferred_none(self, backend): | |||||
pass | |||||
def test_revision_get_ts_none(self, backend): | |||||
pass | |||||
# relation_add tests | |||||
# relation_get tests | |||||
# with_path tests | |||||
def test_with_path(self, backend): | |||||
assert backend.with_path() is True | |||||
# content_find_first tests | |||||
# content_find_all tests |