Commit 9af49a37 authored by Markus Scheidgen's avatar Markus Scheidgen
Browse files

Refactored dataset read functions in coe_repo.

parent 1cdb5401
......@@ -91,7 +91,7 @@ class Calc(Base, datamodel.Calc): # type: ignore
return json.loads(filenames)
@property
def datasets(self) -> List['DataSet']:
def all_datasets(self) -> List['DataSet']:
assert self.calc_id is not None
repo_db = infrastructure.repository_db
query = repo_db.query(literal(self.calc_id).label('calc_id')).cte(recursive=True)
......@@ -102,12 +102,16 @@ class Calc(Base, datamodel.Calc): # type: ignore
query = repo_db.query(query)
dataset_calc_ids = list(r[0] for r in query if not r[0] == self.calc_id)
if len(dataset_calc_ids) > 0:
return list(
return [
DataSet(dataset_calc)
for dataset_calc in repo_db.query(Calc).filter(Calc.calc_id.in_(dataset_calc_ids)))
for dataset_calc in repo_db.query(Calc).filter(Calc.calc_id.in_(dataset_calc_ids))]
else:
return []
@property
def direct_datasets(self) -> List['DataSet']:
return [DataSet(dataset_calc) for dataset_calc in self.parents]
def set_value(self, topic_cid: int, value: str) -> None:
if value is None:
return
......
......@@ -104,27 +104,34 @@ def test_add_upload_metadata(clean_repository_db, processed_upload, other_test_u
assert_coe_upload(processed_upload.upload_hash, empty=empty, meta_data=meta_data)
@pytest.fixture(scope='function')
def datasets(clean_repository_db):
clean_repository_db.begin()
one = Calc()
two = Calc()
three = Calc()
clean_repository_db.add(one)
clean_repository_db.add(two)
clean_repository_db.add(three)
one.children.append(two)
two.children.append(three)
clean_repository_db.commit()
return one, two, three
def test_dataset(datasets):
def ids(id_list):
return [item.id for item in id_list]
one, two, three = datasets
assert len(one.datasets) == 0
assert ids(two.datasets) == [one.calc_id]
assert sorted(ids(three.datasets)) == sorted([one.calc_id, two.calc_id])
class TestDataSets:
@pytest.fixture(scope='function')
def datasets(self, clean_repository_db):
clean_repository_db.begin()
one = Calc()
two = Calc()
three = Calc()
clean_repository_db.add(one)
clean_repository_db.add(two)
clean_repository_db.add(three)
one.children.append(two)
two.children.append(three)
clean_repository_db.commit()
return one, two, three
def assert_datasets(self, datasets, id_list):
assert sorted([ds.id for ds in datasets]) == sorted(id_list)
def test_all(self, datasets):
one, two, three = datasets
self.assert_datasets(one.all_datasets, [])
self.assert_datasets(two.all_datasets, [one.calc_id])
self.assert_datasets(three.all_datasets, [one.calc_id, two.calc_id])
def test_direct(self, datasets):
one, two, three = datasets
self.assert_datasets(one.direct_datasets, [])
self.assert_datasets(two.direct_datasets, [one.calc_id])
self.assert_datasets(three.direct_datasets, [two.calc_id])
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment