Differential D6291 Diff 22866 swh/provenance/tests/mongo/test_backend.py

Changeset 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


				class TestBackend:
				@pytest.fixture
				def backend(self, mongodb: pymongo.database.Database):
				aevisoUnsubmitted Not Done Inline Actions Please add full type annotations (return type and parameters in all methods) aeviso: Please add full type annotations (return type and parameters in all methods)
				return ProvenanceStorageMongoDb(mongodb)

				# add content tests

				def test_add_content_empty(self, backend, mongodb):
				assert backend.content_add({}) is True
				assert mongodb.content.count_documents({}) == 0

				assert backend.content_add([]) is True
				assert mongodb.content.count_documents({}) == 0

				def test_add_content_with_insert(self, backend, mongodb):
				# add data using add_contnet
				# get data from mongo and compare
				sha1 = "cf23df2207d99a74fbe169e3eba035e633b65d94"
				data = {sha1: None}
				assert backend.content_add(data) is True

				assert mongodb.content.count_documents({}) == 1
				cnt = mongodb.content.find_one({"sha1": sha1})
				assert cnt["ts"] is None
				assert cnt["revision"] == {}
				assert cnt["directory"] == {}

				def test_add_content_with_update_later_date(self, backend, mongodb):
				sha1 = "cf23df2207d99a74fbe169e3eba035e633b65d94"
				revision = {"test": "test"}
				mongodb.content.insert_one(
				{"sha1": sha1, "ts": 1631881748, "revision": revision, "directory": {}}
				)

				new_date = datetime(2010, 10, 10)
				data = {sha1: new_date}
				# data has a date before the one in the db
				assert backend.content_add(data) is True
				cnt = mongodb.content.find_one({"sha1": sha1})
				assert cnt["ts"] == new_date.timestamp()
				assert cnt["revision"] == revision
				assert cnt["directory"] == {}

				def test_add_content_with_update_none_date(self, backend, mongodb):
				sha1 = "cf23df2207d99a74fbe169e3eba035e633b65d94"
				revision = {"test": "test"}
				mongodb.content.insert_one(
				{"sha1": sha1, "ts": None, "revision": revision, "directory": {}}
				)

				new_date = datetime(2010, 10, 10)
				data = {sha1: new_date}
				# data has a date before the one in the db
				assert backend.content_add(data) is True
				cnt = mongodb.content.find_one({"sha1": sha1})
				assert cnt["ts"] == new_date.timestamp()
				assert cnt["revision"] == revision
				assert cnt["directory"] == {}

				def test_add_content_do_not_update_older_date(self, backend, mongodb):
				sha1 = "cf23df2207d99a74fbe169e3eba035e633b65d94"
				revision = {"test": "test"}
				mongodb.content.insert_one(
				{"sha1": sha1, "ts": 1286661600, "revision": revision, "directory": {}}
				)

				new_date = datetime(2020, 10, 10)
				data = {sha1: new_date}
				# data has a date after the one in the db
				assert backend.content_add(data) is True
				cnt = mongodb.content.find_one({"sha1": sha1})
				assert cnt["ts"] == 1286661600

				def test_add_content_multiple(self, backend, mongodb):
				pass

				# add directory tests

				def test_add_directory_empty(self, backend, mongodb):
				assert backend.directory_add({}) is True
				assert mongodb.directory.count_documents({}) == 0

				assert backend.directory_add([]) is True
				assert mongodb.directory.count_documents({}) == 0

				def test_add_directory_with_insert(self, backend, mongodb):
				# add data using add_contnet
				# get data from mongo and compare
				sha1 = "cf23df2207d99a74fbe169e3eba035e633b65d94"
				data = {sha1: None}
				assert backend.directory_add(data) is True

				assert mongodb.directory.count_documents({}) == 1
				cnt = mongodb.directory.find_one({"sha1": sha1})
				assert cnt["ts"] is None
				assert cnt["revision"] == {}

				def test_add_directory_with_update_later_date(self, backend, mongodb):
				sha1 = "cf23df2207d99a74fbe169e3eba035e633b65d94"
				revision = {"test": "test"}
				mongodb.directory.insert_one(
				{"sha1": sha1, "ts": 1631881748, "revision": revision}
				)

				new_date = datetime(2010, 10, 10)
				data = {sha1: new_date}
				# data has a date before the one in the db
				assert backend.directory_add(data) is True
				diy = mongodb.directory.find_one({"sha1": sha1})
				assert diy["ts"] == new_date.timestamp()
				assert diy["revision"] == revision

				def test_add_directory_with_update_none_date(self, backend, mongodb):
				sha1 = "cf23df2207d99a74fbe169e3eba035e633b65d94"
				revision = {"test": "test"}
				mongodb.directory.insert_one({"sha1": sha1, "ts": None, "revision": revision})

				new_date = datetime(2010, 10, 10)
				data = {sha1: new_date}
				# data has a date before the one in the db
				assert backend.directory_add(data) is True
				diy = mongodb.directory.find_one({"sha1": sha1})
				assert diy["ts"] == new_date.timestamp()
				assert diy["revision"] == revision

				def test_add_directory_do_not_update_older_date(self, backend, mongodb):
				sha1 = "cf23df2207d99a74fbe169e3eba035e633b65d94"
				revision = {"test": "test"}
				mongodb.directory.insert_one(
				{"sha1": sha1, "ts": 1286661600, "revision": revision}
				)

				new_date = datetime(2020, 10, 10)
				data = {sha1: new_date}
				# data has a date after the one in the db
				assert backend.directory_add(data) is True
				cnt = mongodb.directory.find_one({"sha1": sha1})
				assert cnt["ts"] == 1286661600

				def test_add_directory_multiple(self, backend, mongodb):
				pass