# (c) cavaliba.com - tests / api / pipeline import json import app_home.cache as cache from app_data import crypto from app_data.data import Instance from app_data.loader import load_broker from app_data.models import DataTask from app_data.tasks import submit_pipeline from django.test import TestCase, override_settings from django.urls import reverse from tests import helper def add_pipeline(keyname="test_pipeline", run_permission=None): content = "tasks:\n- field_noop: ['', field1]\n" content_entry = { "classname": "_pipeline", "keyname": keyname, "displayname": keyname, "is_enabled": True, "content": content, } if run_permission: content_entry["run_permission"] = run_permission aaa = {"perms": ["p_pipeline_create"]} load_broker(datalist=[content_entry], aaa=aaa) def add_apikey_noperm(keyname="test_noperm", secret="noperm_secret"): hashed = crypto.hash_create(secret) instance = Instance(classname="_apikey", keyname=keyname) data = { "secret": hashed, "ip_filter": "*", "is_readonly": False, "is_enabled": True, "acl": "role:role_nonexistent", } instance.merge_import(data) instance.create() return {"X-Cavaliba-Key": f"{keyname} {secret}"} def add_apikey_readonly(keyname="test_readonly", secret="readonly_secret"): hashed = crypto.hash_create(secret) instance = Instance(classname="_apikey", keyname=keyname) data = { "secret": hashed, "ip_filter": "*", "is_readonly": True, "is_enabled": True, "acl": "role:role_admin", } instance.merge_import(data) instance.create() return {"X-Cavaliba-Key": f"{keyname} {secret}"} class APIPipelineAuthTest(TestCase): fixtures = ["test"] def setUp(self): cache.clear() self.header = helper.add_apikey_admin() add_pipeline() helper.add_schema(classname="ptest") def test_api_pipeline_no_auth(self): response = self.client.post( reverse('api:api_pipeline', kwargs={'key': 'test_pipeline'}) + '?schema=ptest', ) self.assertEqual(response.status_code, 401) def test_api_pipeline_no_permission(self): header = add_apikey_noperm() response = self.client.post( reverse('api:api_pipeline', kwargs={'key': 'test_pipeline'}) + '?schema=ptest', headers=header, ) self.assertEqual(response.status_code, 401) def test_api_pipeline_readonly_key(self): header = add_apikey_readonly() response = self.client.post( reverse('api:api_pipeline', kwargs={'key': 'test_pipeline'}) + '?schema=ptest', headers=header, ) self.assertEqual(response.status_code, 401) def test_api_pipeline_get_not_allowed(self): response = self.client.get( reverse('api:api_pipeline', kwargs={'key': 'test_pipeline'}) + '?schema=ptest', headers=self.header, ) self.assertEqual(response.status_code, 405) class APIPipelineInputTest(TestCase): fixtures = ["test"] def setUp(self): cache.clear() self.header = helper.add_apikey_admin() add_pipeline() helper.add_schema(classname="ptest") def test_api_pipeline_missing_schema(self): response = self.client.post( reverse('api:api_pipeline', kwargs={'key': 'test_pipeline'}), headers=self.header, ) self.assertEqual(response.status_code, 400) def test_api_pipeline_not_found_by_name(self): response = self.client.post( reverse('api:api_pipeline', kwargs={'key': 'no_such_pipeline'}) + '?schema=ptest', headers=self.header, ) self.assertEqual(response.status_code, 404) def test_api_pipeline_not_found_by_id(self): response = self.client.post( reverse('api:api_pipeline', kwargs={'id': 999999}) + '?schema=ptest', headers=self.header, ) self.assertEqual(response.status_code, 404) @override_settings(CELERY_TASK_ALWAYS_EAGER=True) class APIPipelineResponseTest(TestCase): """Test the 202 async response shape from the API.""" fixtures = ["test"] def setUp(self): cache.clear() self.header = helper.add_apikey_admin() add_pipeline() helper.add_schema(classname="ptest") helper.add_instance(classname="ptest", keyname="obj01") helper.add_instance(classname="ptest", keyname="obj02") def test_api_pipeline_returns_202(self): response = self.client.post( reverse('api:api_pipeline', kwargs={'key': 'test_pipeline'}) + '?schema=ptest', headers=self.header, ) self.assertEqual(response.status_code, 202) def test_api_pipeline_response_has_handle_and_task_url(self): response = self.client.post( reverse('api:api_pipeline', kwargs={'key': 'test_pipeline'}) + '?schema=ptest', headers=self.header, ) self.assertEqual(response.status_code, 202) data = json.loads(response.content) self.assertIn("pipeline", data) self.assertIn("dryrun", data) self.assertIn("handle", data) self.assertIn("task_url", data) self.assertEqual(data["pipeline"], "test_pipeline") def test_api_pipeline_by_id_returns_202(self): inst = Instance.from_keyname(classname="_pipeline", keyname="test_pipeline") response = self.client.post( reverse('api:api_pipeline', kwargs={'id': inst.id}) + '?schema=ptest', headers=self.header, ) self.assertEqual(response.status_code, 202) data = json.loads(response.content) self.assertIn("handle", data) self.assertEqual(data["pipeline"], "test_pipeline") class APIPipelineLogicTest(TestCase): """Test pipeline execution results using sync=True via submit_pipeline.""" fixtures = ["test"] def setUp(self): cache.clear() helper.add_admin_user(login="unittest") add_pipeline() helper.add_schema(classname="ptest") helper.add_instance(classname="ptest", keyname="obj01") helper.add_instance(classname="ptest", keyname="obj02") def test_pipeline_run_by_name_dryrun(self): aaa = {'perms': ['p_data_admin', 'p_pipeline_run']} count_ok, count_discarded, errors = helper.run_pipeline("test_pipeline", ["ptest"], aaa=aaa, dryrun=True) self.assertEqual(errors, []) self.assertEqual(count_discarded, 0) self.assertEqual(count_ok, 2) def test_pipeline_run_dryrun_false(self): aaa = {'perms': ['p_data_admin', 'p_pipeline_run']} count_ok, count_discarded, errors = helper.run_pipeline("test_pipeline", ["ptest"], aaa=aaa, dryrun=False) self.assertEqual(errors, []) self.assertEqual(count_ok, 2) def test_pipeline_multi_schema(self): helper.add_schema(classname="ptest2") helper.add_instance(classname="ptest2", keyname="obj03") aaa = {'perms': ['p_data_admin', 'p_pipeline_run']} count_ok, count_discarded, errors = helper.run_pipeline("test_pipeline", ["ptest", "ptest2"], aaa=aaa, dryrun=True) self.assertEqual(errors, []) self.assertEqual(count_ok, 3) def test_pipeline_user_schema(self): aaa = {'perms': ['p_data_admin', 'p_pipeline_run', 'p_user_update']} handle, err = submit_pipeline("test_pipeline", ["_user"], dryrun=True, aaa=aaa, sync=True) self.assertIsNotNone(handle) dt = DataTask.objects.get(handle=handle) self.assertEqual(dt.state, "DONE") output = dt.output or {} self.assertEqual(output["results"][0]["schema"], "_user") def test_pipeline_task_output_structure(self): aaa = {'perms': ['p_data_admin', 'p_pipeline_run']} handle, err = submit_pipeline("test_pipeline", ["ptest"], dryrun=True, aaa=aaa, sync=True) self.assertIsNotNone(handle) dt = DataTask.objects.get(handle=handle) self.assertEqual(dt.state, "DONE") output = dt.output or {} self.assertIn("pipeline", output) self.assertIn("dryrun", output) self.assertIn("total_ok", output) self.assertIn("total_discarded", output) self.assertIn("total_errors", output) self.assertIn("results", output) self.assertEqual(len(output["results"]), 1) result = output["results"][0] self.assertIn("schema", result) self.assertIn("count_ok", result) self.assertIn("count_discarded", result) self.assertIn("errors", result)