Page MenuHomeSoftware Heritage
Differential D509 Diff 1585 swh/storage/tests/test_storage.py

Changeset View

Standalone View

View Options

swh/storage/tests/test_storage.py

# Copyright (C) 2015-2017 The Software Heritage developers # Copyright (C) 2015-2017 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution # See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version # License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information # See top-level LICENSE file for more information
from collections import defaultdict from collections import defaultdict
import copy import copy
import datetime import datetime
from operator import itemgetter from operator import itemgetter
import psycopg2 import psycopg2
import unittest import unittest
from uuid import UUID
from unittest.mock import Mock, patch from unittest.mock import Mock, patch
from nose.tools import istest from nose.tools import istest
from nose.plugins.attrib import attr from nose.plugins.attrib import attr
from swh.model import from_disk, identifiers from swh.model import from_disk, identifiers
from swh.model.hashutil import hash_to_bytes from swh.model.hashutil import hash_to_bytes
from swh.core.tests.db_testing import DbTestFixture from swh.core.tests.db_testing import DbTestFixture
▲ Show 20 Lines • Show All 450 Lines • ▼ Show 20 Lines def setUp(self):
self.fetch_history_data = { self.fetch_history_data = {
'status': True, 'status': True,
'result': {'foo': 'bar'}, 'result': {'foo': 'bar'},
'stdout': 'blabla', 'stdout': 'blabla',
'stderr': 'blablabla', 'stderr': 'blablabla',
} }
self.entity1 = {
'uuid': UUID('f96a7ec1-0058-4920-90cc-7327e4b5a4bf'),
# GitHub users
'parent': UUID('ad6df473-c1d2-4f40-bc58-2b091d4a750e'),
'name': 'github:user:olasd',
'type': 'person',
'description': 'Nicolas Dandrimont',
'homepage': 'http://example.com',
'active': True,
'generated': True,
'lister_metadata': {
# swh.lister.github
'lister': '34bd6b1b-463f-43e5-a697-785107f598e4',
'id': 12877,
'type': 'user',
'last_activity': '2015-11-03',
},
'metadata': None,
'validity': [
datetime.datetime(2015, 11, 3, 11, 0, 0,
tzinfo=datetime.timezone.utc),
]
}
self.entity1_query = {
'lister': '34bd6b1b-463f-43e5-a697-785107f598e4',
'id': 12877,
'type': 'user',
}
self.entity2 = {
'uuid': UUID('3903d075-32d6-46d4-9e29-0aef3612c4eb'),
# GitHub users
'parent': UUID('ad6df473-c1d2-4f40-bc58-2b091d4a750e'),
'name': 'github:user:zacchiro',
'type': 'person',
'description': 'Stefano Zacchiroli',
'homepage': 'http://example.com',
'active': True,
'generated': True,
'lister_metadata': {
# swh.lister.github
'lister': '34bd6b1b-463f-43e5-a697-785107f598e4',
'id': 216766,
'type': 'user',
'last_activity': '2015-11-03',
},
'metadata': None,
'validity': [
datetime.datetime(2015, 11, 3, 11, 0, 0,
tzinfo=datetime.timezone.utc),
]
}
self.entity3 = {
'uuid': UUID('111df473-c1d2-4f40-bc58-2b091d4a7111'),
# GitHub users
'parent': UUID('222df473-c1d2-4f40-bc58-2b091d4a7222'),
'name': 'github:user:ardumont',
'type': 'person',
'description': 'Antoine R. Dumont a.k.a tony',
'homepage': 'https://ardumont.github.io',
'active': True,
'generated': True,
'lister_metadata': {
'lister': '34bd6b1b-463f-43e5-a697-785107f598e4',
'id': 666,
'type': 'user',
'last_activity': '2016-01-15',
},
'metadata': None,
'validity': [
datetime.datetime(2015, 11, 3, 11, 0, 0,
tzinfo=datetime.timezone.utc),
]
}
self.entity4 = {
'uuid': UUID('222df473-c1d2-4f40-bc58-2b091d4a7222'),
# GitHub users
'parent': None,
'name': 'github:user:ToNyX',
'type': 'person',
'description': 'ToNyX',
'homepage': 'https://ToNyX.github.io',
'active': True,
'generated': True,
'lister_metadata': {
'lister': '34bd6b1b-463f-43e5-a697-785107f598e4',
'id': 999,
'type': 'user',
'last_activity': '2015-12-24',
},
'metadata': None,
'validity': [
datetime.datetime(2015, 11, 3, 11, 0, 0,
tzinfo=datetime.timezone.utc),
]
}
self.entity2_query = {
'lister_metadata': {
'lister': '34bd6b1b-463f-43e5-a697-785107f598e4',
'id': 216766,
'type': 'user',
},
}
self.snapshot = { self.snapshot = {
'id': hash_to_bytes('2498dbf535f882bc7f9a18fb16c9ad27fda7bab7'), 'id': hash_to_bytes('2498dbf535f882bc7f9a18fb16c9ad27fda7bab7'),
'branches': { 'branches': {
self.occurrence['branch']: { self.occurrence['branch']: {
'target': self.occurrence['target'], 'target': self.occurrence['target'],
'target_type': self.occurrence['target_type'], 'target_type': self.occurrence['target_type'],
}, },
}, },
▲ Show 20 Lines • Show All 697 Lines • ▼ Show 20 Lines def origin_get(self):
'type': self.origin['type']}) 'type': self.origin['type']})
self.assertEqual(actual_origin0['id'], id) self.assertEqual(actual_origin0['id'], id)
# lookup per id (returns dict) # lookup per id (returns dict)
actual_origin1 = self.storage.origin_get({'id': id}) actual_origin1 = self.storage.origin_get({'id': id})
self.assertEqual(actual_origin1, {'id': id, self.assertEqual(actual_origin1, {'id': id,
'type': self.origin['type'], 'type': self.origin['type'],
'url': self.origin['url'], 'url': self.origin['url']})
'lister': None,
'project': None})
@istest @istest
def origin_search(self): def origin_search(self):
found_origins = list(self.storage.origin_search(self.origin['url'])) found_origins = list(self.storage.origin_search(self.origin['url']))
self.assertEqual(len(found_origins), 0) self.assertEqual(len(found_origins), 0)
found_origins = list(self.storage.origin_search(self.origin['url'], found_origins = list(self.storage.origin_search(self.origin['url'],
regexp=True)) regexp=True))
self.assertEqual(len(found_origins), 0) self.assertEqual(len(found_origins), 0)
id = self.storage.origin_add_one(self.origin) id = self.storage.origin_add_one(self.origin)
origin_data = {'id': id, origin_data = {'id': id,
'type': self.origin['type'], 'type': self.origin['type'],
'url': self.origin['url'], 'url': self.origin['url']}
'lister': None,
'project': None}
found_origins = list(self.storage.origin_search(self.origin['url'])) found_origins = list(self.storage.origin_search(self.origin['url']))
self.assertEqual(len(found_origins), 1) self.assertEqual(len(found_origins), 1)
self.assertEqual(found_origins[0], origin_data) self.assertEqual(found_origins[0], origin_data)
found_origins = list(self.storage.origin_search( found_origins = list(self.storage.origin_search(
'.' + self.origin['url'][1:-1] + '.', regexp=True)) '.' + self.origin['url'][1:-1] + '.', regexp=True))
self.assertEqual(len(found_origins), 1) self.assertEqual(len(found_origins), 1)
self.assertEqual(found_origins[0], origin_data) self.assertEqual(found_origins[0], origin_data)
id2 = self.storage.origin_add_one(self.origin2) id2 = self.storage.origin_add_one(self.origin2)
origin2_data = {'id': id2, origin2_data = {'id': id2,
'type': self.origin2['type'], 'type': self.origin2['type'],
'url': self.origin2['url'], 'url': self.origin2['url']}
'lister': None,
'project': None}
found_origins = list(self.storage.origin_search(self.origin2['url'])) found_origins = list(self.storage.origin_search(self.origin2['url']))
self.assertEqual(len(found_origins), 1) self.assertEqual(len(found_origins), 1)
self.assertEqual(found_origins[0], origin2_data) self.assertEqual(found_origins[0], origin2_data)
found_origins = list(self.storage.origin_search( found_origins = list(self.storage.origin_search(
'.' + self.origin2['url'][1:-1] + '.', regexp=True)) '.' + self.origin2['url'][1:-1] + '.', regexp=True))
self.assertEqual(len(found_origins), 1) self.assertEqual(len(found_origins), 1)
self.assertEqual(found_origins[0], origin2_data) self.assertEqual(found_origins[0], origin2_data)
▲ Show 20 Lines • Show All 466 Lines • ▼ Show 20 Lines def snapshot_get_latest(self):
# Check that the status filter is still working # Check that the status filter is still working
self.assertEquals( self.assertEquals(
self.complete_snapshot, self.complete_snapshot,
self.storage.snapshot_get_latest(origin_id, self.storage.snapshot_get_latest(origin_id,
allowed_statuses=['full']), allowed_statuses=['full']),
) )
@istest @istest
def entity_get_from_lister_metadata(self):
self.storage.entity_add([self.entity1])
fetched_entities = list(
self.storage.entity_get_from_lister_metadata(
[self.entity1_query, self.entity2_query]))
# Entity 1 should have full metadata, with last_seen/last_id instead
# of validity
entity1 = self.entity1.copy()
entity1['last_seen'] = entity1['validity'][0]
del fetched_entities[0]['last_id']
del entity1['validity']
# Entity 2 should have no metadata
entity2 = {
'uuid': None,
'lister_metadata': self.entity2_query.copy(),
}
self.assertEquals(fetched_entities, [entity1, entity2])
@istest
def entity_get_from_lister_metadata_twice(self):
self.storage.entity_add([self.entity1])
fetched_entities1 = list(
self.storage.entity_get_from_lister_metadata(
[self.entity1_query]))
fetched_entities2 = list(
self.storage.entity_get_from_lister_metadata(
[self.entity1_query]))
self.assertEquals(fetched_entities1, fetched_entities2)
@istest
def entity_get(self):
# given
self.storage.entity_add([self.entity4])
self.storage.entity_add([self.entity3])
# when: entity3 -child-of-> entity4
actual_entity3 = list(self.storage.entity_get(self.entity3['uuid']))
self.assertEquals(len(actual_entity3), 2)
# remove dynamic data (modified by db)
entity3 = self.entity3.copy()
entity4 = self.entity4.copy()
del entity3['validity']
del entity4['validity']
del actual_entity3[0]['last_seen']
del actual_entity3[0]['last_id']
del actual_entity3[1]['last_seen']
del actual_entity3[1]['last_id']
self.assertEquals(actual_entity3, [entity3, entity4])
# when: entity4 only child
actual_entity4 = list(self.storage.entity_get(self.entity4['uuid']))
self.assertEquals(len(actual_entity4), 1)
# remove dynamic data (modified by db)
entity4 = self.entity4.copy()
del entity4['validity']
del actual_entity4[0]['last_id']
del actual_entity4[0]['last_seen']
self.assertEquals(actual_entity4, [entity4])
@istest
def entity_get_one(self):
# given
self.storage.entity_add([self.entity3, self.entity4])
# when: entity3 -child-of-> entity4
actual_entity3 = self.storage.entity_get_one(self.entity3['uuid'])
# remove dynamic data (modified by db)
entity3 = self.entity3.copy()
del entity3['validity']
del actual_entity3['last_seen']
del actual_entity3['last_id']
self.assertEquals(actual_entity3, entity3)
@istest
def stat_counters(self): def stat_counters(self):
expected_keys = ['content', 'directory', 'directory_entry_dir', expected_keys = ['content', 'directory', 'directory_entry_dir',
'origin', 'person', 'revision'] 'origin', 'person', 'revision']
for key in expected_keys: for key in expected_keys:
self.cursor.execute('select * from swh_update_counter(%s)', (key,)) self.cursor.execute('select * from swh_update_counter(%s)', (key,))
self.conn.commit() self.conn.commit()
▲ Show 20 Lines • Show All 594 Lines • Show Last 20 Lines
About · Developer information · Legal