@pytest.fixture
def do_action(data_catalog, admin_client, deep_file_tree):
def _do(
state,
has_files,
action=None,
cumulative_state=Dataset.CumulativeState.NOT_CUMULATIVE,
patch_fields={},
):
if state == "published" or state == "draft_of":
dataset = factories.PublishedDatasetFactory(
data_catalog=data_catalog, cumulative_state=cumulative_state
)
elif state == "draft":
dataset = factories.DatasetFactory(
data_catalog=data_catalog, cumulative_state=cumulative_state
)
if has_files:
initial_actions = {
**deep_file_tree["params"],
"directory_actions": [{"pathname": "/dir1/"}],
}
res = admin_client.patch(
f"/v3/datasets/{dataset.id}",
{"fileset": initial_actions},
content_type="application/json",
)
assert res.status_code == 200
dataset.refresh_from_db()
if state == "draft_of":
dataset = dataset.create_new_draft()
if action:
actions = {
**deep_file_tree["params"],
"directory_actions": [
{"pathname": "/dir1/", "action": action},
{"pathname": "/dir2/", "action": action},
],
}
res = admin_client.patch(
f"/v3/datasets/{dataset.id}",
{"fileset": actions, **patch_fields},
content_type="application/json",
)
else:
res = admin_client.patch(
f"/v3/datasets/{dataset.id}",
patch_fields,
content_type="application/json",
)
return res
return _do