Organization Validation for Service Inventory by JVickery-TBS · Pull Request #1668 · open-data/ckanext-canada

JVickery-TBS · 2026-04-09T20:13:31Z

We are able to consistently use the releases from the SI github to generate the CSV ref data we need. We have commands which load it into the tables which we define in an sql script in here. The part we were tripped on was relation between Service IDs and Program IDs, but there is no relation that is being defined or maintained currently so don't have to worry about that.

Just extended the temp datastore table to add an optional org_name to it, which we then use in the triggers. The Service ID one works fine, need to make the error message for it still.

The Program IDs one is gunna be interesting as Program IDs is a multiple select.

- Started working on reference tables for pd data.

- Continued script to generate service reference data. - Continued schema for ref tables. - Improved pd subcommand for loading ref data.

- Finalized script to generate service ref data. - Added more recombinant schema key/values.

- Made JSONB index. - Added choices_filter_query. - Added recombinant org_name to the temp ds table. - Started writing the database validation.

- Finalized ref value checks and errors in the db func. - Fixed up some other logic things.

- Started working on fiscal year function.

- Finalized fiscal year db function.

wardi · 2026-04-18T15:54:29Z

just those minor things otherwise LGTM

- Added change log file.

- Pyright fixes. - Flake8 fixes.

- Made service name fields computed ones. - Modified filter scripts for service name lookups.

- Filter script typo.

codecov · 2026-04-20T20:20:29Z

❌ 19 Tests Failed:

Tests completed	Failed	Passed	Skipped
276	19	257	0

View the top 3 failed test(s) by shortest run time

ckanext/canada/tests/test_service.py::TestStdService::test_integers

Stack Traces | 0.026s run time

self = <ckanext.canada.tests.test_service.TestStdService object at 0x7f54336cd4f0>

    def test_integers(self):
        """
        Range and positive integer validation
        """
>       self._make_parent_record()

.../canada/tests/test_service.py:573: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../canada/tests/test_service.py:370: in _make_parent_record
    self.lc.action.datastore_upsert(
.../ckanapi/ckanapi/common.py:60: in action
    return self._ckan.call_action(name, data_dict=kwargs)
.../ckanapi/ckanapi/localckan.py:74: in call_action
    return self._get_action(action)(context, data_dict)
.../ckan/logic/__init__.py:581: in wrapped
    result = _action(context, data_dict, **kw)
.../ckanext/recombinant/logic.py:517: in recombinant_datastore_upsert
    return up_func(context, data_dict)
.../datastore/logic/action.py:296: in datastore_upsert
    result = backend.upsert(context, data_dict)
.../datastore/backend/postgres.py:2575: in upsert
    return upsert(context, data_dict)
.../datastore/backend/postgres.py:2061: in upsert
    upsert_data(dict(context, connection=connection), data_dict)
.../canada/plugin/internal_plugin.py:82: in patched_upsert_data
    raise e
.../canada/plugin/internal_plugin.py:69: in patched_upsert_data
    return original_upsert_data(context, data_dict)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

context = {'__auth_audit': [], '__auth_user_obj_checked': True, 'auth_user_obj': <User id=bb096f29-9a92-47a7-8e07-1241c6da4624 n...ve image_url=None plugin_extras=None>, 'connection': <sqlalchemy.engine.base.Connection object at 0x7f543335e400>, ...}
data_dict = {'connection_url': 'postgresql://datastore_write:pass@postgres/datastore_test', 'records': [{'automated_decision_syste...channel': ['EML', 'FAX', 'PERSON', 'ONL', 'POST', 'TEL'], ...}], 'resource_id': '24b243c8-709d-48ef-9a0f-7dcab86f3a34'}

    def upsert_data(context: Context, data_dict: dict[str, Any]):
        '''insert all data from records'''
        if not data_dict.get('records'):
            return
    
        method = data_dict.get('method', _UPSERT)
    
        fields = _get_fields(context['connection'], data_dict['resource_id'])
        field_names = _pluck('id', fields)
        records = data_dict['records']
        sql_columns = ", ".join(
            identifier(name) for name in field_names)
        num = -1
    
        if method == _INSERT:
            rows = []
            for num, record in enumerate(records):
                _validate_record(record, num, field_names)
    
                row = {}
                for idx, field in enumerate(fields):
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        value = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        value = None
                    row[f"val_{idx}"] = value
                rows.append(row)
    
            sql_string = '''INSERT INTO {res_id} ({columns})
                VALUES ({values});'''.format(
                res_id=identifier(data_dict['resource_id']),
                columns=sql_columns,
                values=', '.join([
                    f":val_{idx}" for idx in range(0, len(field_names))
                ])
            )
    
            try:
                context['connection'].execute(sa.text(sql_string), rows)
            except (DatabaseError, DataError) as err:
                # (canada fork only): parse constraint sql errors
                # TODO: upstream contrib!!
                errmsg = _programming_error_summary(err)
                if 'violates foreign key constraint' in errmsg:
                    _ = lambda x:x
                    errmsg = _(
                        'Cannot insert records ({refValues}) because '\
                        'they do not exist in the referenced table. '\
                        'Referencing {refKeys} from {refTable}.')
                    raise ValidationError(dict(
                        _parse_constraint_error_from_psql_error(err, errmsg),
                        records_row=num))
                raise ValidationError({
                    'records': [errmsg],
                    'records_row': num,
                })
    
        elif method in [_UPDATE, _UPSERT]:
            unique_keys = _get_unique_key(context, data_dict)
    
            for num, record in enumerate(records):
                if not unique_keys and '_id' not in record:
                    raise ValidationError({
                        'table': [u'unique key must be passed for update/upsert']
                    })
    
                elif '_id' not in record:
                    # all key columns have to be defined
                    missing_fields = [field for field in unique_keys
                                      if field not in record]
                    if missing_fields:
                        raise ValidationError({
                            'key': [u'''fields "{fields}" are missing
                                but needed as key'''.format(
                                    fields=', '.join(missing_fields))]
                        })
    
                for field in fields:
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        record[field['id']] = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        record[field['id']] = None
    
                non_existing_field_names = [
                    field for field in record
                    if field not in field_names and field != '_id'
                ]
                if non_existing_field_names:
>                   raise ValidationError({
                        'fields': [u'fields "{0}" do not exist'.format(
                            ', '.join(non_existing_field_names))]
                    })
E                   ckan.logic.ValidationError: None - {'fields': ['fields "service_name_en, service_name_fr" do not exist']}

.../datastore/backend/postgres.py:1577: ValidationError

ckanext/canada/tests/test_service.py::TestService::test_max_chars

Stack Traces | 0.027s run time

self = <ckanext.canada.tests.test_service.TestService object at 0x7feca7726d30>

    def test_max_chars(self):
        """
        Over max character field values should raise an exception
        """
        chromo = get_chromo('service')
        record = chromo['examples']['record'].copy()
    
        expect_maxchar_fields = ['service_name_en', 'service_name_fr',
                                 'service_description_en', 'service_description_fr',
                                 'automated_decision_system_description_en',
                                 'automated_decision_system_description_fr',
                                 'os_comments_client_interaction_en',
                                 'os_comments_client_interaction_fr',
                                 'special_remarks_en', 'special_remarks_fr',
                                 'service_uri_en', 'service_uri_fr']
    
        for field in chromo['fields']:
            if field.get('max_chars'):
                assert field['datastore_id'] in expect_maxchar_fields
                record[field['datastore_id']] = 'xx' * field.get('max_chars')
    
        with pytest.raises(ValidationError) as ve:
            self.lc.action.datastore_upsert(
                resource_id=self.resource_id,
                records=[record])
        model.Session.rollback()
        err = ve.value.error_dict
>       assert 'records' in err
E       assert 'records' in {'fields': ['fields "service_name_en, service_name_fr" do not exist']}

.../canada/tests/test_service.py:306: AssertionError

ckanext/canada/tests/test_service.py::TestStdService::test_choice_fields

Stack Traces | 0.027s run time

self = <ckanext.canada.tests.test_service.TestStdService object at 0x7f5433fa94c0>

    def test_choice_fields(self):
        """
        Fields with choices should expect those values
        """
        chromo = get_chromo('service-std')
        record = chromo['examples']['record'].copy()
    
        expected_choice_fields = ['fiscal_yr', 'type', 'channel']
    
        for field in chromo['fields']:
            if field.get('published_resource_computed_field'):
                continue
            if 'choices_file' in field or 'choices' in field:
                assert field['datastore_id'] in expected_choice_fields
                if field['datastore_type'] == '_text':
                    record[field['datastore_id']] = ['zzz']
                else:
                    record[field['datastore_id']] = 'zzz'
    
        with pytest.raises(ValidationError) as ve:
            self.lc.action.datastore_upsert(
                resource_id=self.resource_id,
                records=[record])
        model.Session.rollback()
        err = ve.value.error_dict
>       assert 'records' in err
E       assert 'records' in {'fields': ['fields "service_name_en, service_name_fr" do not exist']}

.../canada/tests/test_service.py:533: AssertionError

ckanext/canada/tests/test_service.py::TestService::test_int_na_nd

Stack Traces | 0.03s run time

self = <ckanext.canada.tests.test_service.TestService object at 0x7f7f32f341c0>

    def test_int_na_nd(self):
        """
        Special ND, NA, or positive integer fields
        """
        record = get_chromo('service')['examples']['record'].copy()
    
        expected_int_na_nd_fields = ['num_phone_enquiries', 'num_applications_by_phone',
                                     'num_website_visits', 'num_applications_online',
                                     'num_applications_in_person',
                                     'num_applications_by_mail',
                                     'num_applications_by_email',
                                     'num_applications_by_fax',
                                     'num_applications_by_other']
    
        for int_na_nd_field in expected_int_na_nd_fields:
            record[int_na_nd_field] = 'BLOOP'
    
        with pytest.raises(ValidationError) as ve:
            self.lc.action.datastore_upsert(
                resource_id=self.resource_id,
                records=[record])
        model.Session.rollback()
        err = ve.value.error_dict
>       assert 'records' in err
E       assert 'records' in {'fields': ['fields "service_name_en, service_name_fr" do not exist']}

.../canada/tests/test_service.py:333: AssertionError

ckanext/canada/tests/test_service.py::TestStdService::test_max_chars

Stack Traces | 0.032s run time

self = <ckanext.canada.tests.test_service.TestStdService object at 0x7efe929d69a0>

    def test_max_chars(self):
        """
        Over max character field values should raise an exception
        """
>       self._make_parent_record()

.../canada/tests/test_service.py:541: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../canada/tests/test_service.py:370: in _make_parent_record
    self.lc.action.datastore_upsert(
.../ckanapi/ckanapi/common.py:60: in action
    return self._ckan.call_action(name, data_dict=kwargs)
.../ckanapi/ckanapi/localckan.py:74: in call_action
    return self._get_action(action)(context, data_dict)
.../ckan/logic/__init__.py:581: in wrapped
    result = _action(context, data_dict, **kw)
.../ckanext/recombinant/logic.py:517: in recombinant_datastore_upsert
    return up_func(context, data_dict)
.../datastore/logic/action.py:296: in datastore_upsert
    result = backend.upsert(context, data_dict)
.../datastore/backend/postgres.py:2575: in upsert
    return upsert(context, data_dict)
.../datastore/backend/postgres.py:2061: in upsert
    upsert_data(dict(context, connection=connection), data_dict)
.../canada/plugin/internal_plugin.py:82: in patched_upsert_data
    raise e
.../canada/plugin/internal_plugin.py:69: in patched_upsert_data
    return original_upsert_data(context, data_dict)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

context = {'__auth_audit': [], '__auth_user_obj_checked': True, 'auth_user_obj': <User id=78ff605d-85a2-48ee-b95b-fa96376fb533 n...ve image_url=None plugin_extras=None>, 'connection': <sqlalchemy.engine.base.Connection object at 0x7efe8fdb2c70>, ...}
data_dict = {'connection_url': 'postgresql://datastore_write:pass@postgres/datastore_test', 'records': [{'automated_decision_syste...channel': ['EML', 'FAX', 'PERSON', 'ONL', 'POST', 'TEL'], ...}], 'resource_id': '34ce13d0-5255-4951-a6f4-f4512850b99f'}

    def upsert_data(context: Context, data_dict: dict[str, Any]):
        '''insert all data from records'''
        if not data_dict.get('records'):
            return
    
        method = data_dict.get('method', _UPSERT)
    
        fields = _get_fields(context['connection'], data_dict['resource_id'])
        field_names = _pluck('id', fields)
        records = data_dict['records']
        sql_columns = ", ".join(
            identifier(name) for name in field_names)
        num = -1
    
        if method == _INSERT:
            rows = []
            for num, record in enumerate(records):
                _validate_record(record, num, field_names)
    
                row = {}
                for idx, field in enumerate(fields):
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        value = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        value = None
                    row[f"val_{idx}"] = value
                rows.append(row)
    
            sql_string = '''INSERT INTO {res_id} ({columns})
                VALUES ({values});'''.format(
                res_id=identifier(data_dict['resource_id']),
                columns=sql_columns,
                values=', '.join([
                    f":val_{idx}" for idx in range(0, len(field_names))
                ])
            )
    
            try:
                context['connection'].execute(sa.text(sql_string), rows)
            except (DatabaseError, DataError) as err:
                # (canada fork only): parse constraint sql errors
                # TODO: upstream contrib!!
                errmsg = _programming_error_summary(err)
                if 'violates foreign key constraint' in errmsg:
                    _ = lambda x:x
                    errmsg = _(
                        'Cannot insert records ({refValues}) because '\
                        'they do not exist in the referenced table. '\
                        'Referencing {refKeys} from {refTable}.')
                    raise ValidationError(dict(
                        _parse_constraint_error_from_psql_error(err, errmsg),
                        records_row=num))
                raise ValidationError({
                    'records': [errmsg],
                    'records_row': num,
                })
    
        elif method in [_UPDATE, _UPSERT]:
            unique_keys = _get_unique_key(context, data_dict)
    
            for num, record in enumerate(records):
                if not unique_keys and '_id' not in record:
                    raise ValidationError({
                        'table': [u'unique key must be passed for update/upsert']
                    })
    
                elif '_id' not in record:
                    # all key columns have to be defined
                    missing_fields = [field for field in unique_keys
                                      if field not in record]
                    if missing_fields:
                        raise ValidationError({
                            'key': [u'''fields "{fields}" are missing
                                but needed as key'''.format(
                                    fields=', '.join(missing_fields))]
                        })
    
                for field in fields:
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        record[field['id']] = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        record[field['id']] = None
    
                non_existing_field_names = [
                    field for field in record
                    if field not in field_names and field != '_id'
                ]
                if non_existing_field_names:
>                   raise ValidationError({
                        'fields': [u'fields "{0}" do not exist'.format(
                            ', '.join(non_existing_field_names))]
                    })
E                   ckan.logic.ValidationError: None - {'fields': ['fields "service_name_en, service_name_fr" do not exist']}

.../datastore/backend/postgres.py:1577: ValidationError

ckanext/canada/tests/test_service.py::TestStdService::test_foreign_constraint

Stack Traces | 0.479s run time

self = <ckanext.canada.tests.test_service.TestStdService object at 0x7fb7dd21f640>

    def test_foreign_constraint(self):
        """
        Trying to create a Standard record referencing a nonexistent Service record should raise an exception
        """
        record = get_chromo('service-std')['examples']['record'].copy()
        record['service_id'] = '1002_does_not_exist'
        with pytest.raises(ValidationError) as ve:
            self.lc.action.datastore_upsert(
                resource_id=self.resource_id,
                records=[record])
        model.Session.rollback()
        err = ve.value.error_dict
>       assert 'constraint_info' in err
E       assert 'constraint_info' in {'fields': ['fields "service_name_en, service_name_fr" do not exist']}

.../canada/tests/test_service.py:415: AssertionError

ckanext/canada/tests/test_service.py::TestService::test_primary_key_commas

Stack Traces | 0.549s run time

self = <ckanext.canada.tests.test_service.TestService object at 0x7fadce14e3d0>

    def test_primary_key_commas(self):
        """
        Commas in primary keys should error
        """
        record = get_chromo('service')['examples']['record'].copy()
        record['service_id'] = 'this,is,a,failure'
        with pytest.raises(ValidationError) as ve:
            self.lc.action.datastore_upsert(
                resource_id=self.resource_id,
                records=[record])
        model.Session.rollback()
        err = ve.value.error_dict
>       assert 'records' in err
E       assert 'records' in {'fields': ['fields "service_name_en, service_name_fr" do not exist']}

.../canada/tests/test_service.py:51: AssertionError

ckanext/canada/tests/test_service.py::TestStdService::test_required_fields

Stack Traces | 0.562s run time

self = <ckanext.canada.tests.test_service.TestStdService object at 0x7f54336cd460>

    def test_required_fields(self):
        """
        Excluding required fields should raise an exception
        """
>       self._make_parent_record()

.../canada/tests/test_service.py:435: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../canada/tests/test_service.py:370: in _make_parent_record
    self.lc.action.datastore_upsert(
.../ckanapi/ckanapi/common.py:60: in action
    return self._ckan.call_action(name, data_dict=kwargs)
.../ckanapi/ckanapi/localckan.py:74: in call_action
    return self._get_action(action)(context, data_dict)
.../ckan/logic/__init__.py:581: in wrapped
    result = _action(context, data_dict, **kw)
.../ckanext/recombinant/logic.py:517: in recombinant_datastore_upsert
    return up_func(context, data_dict)
.../datastore/logic/action.py:296: in datastore_upsert
    result = backend.upsert(context, data_dict)
.../datastore/backend/postgres.py:2575: in upsert
    return upsert(context, data_dict)
.../datastore/backend/postgres.py:2061: in upsert
    upsert_data(dict(context, connection=connection), data_dict)
.../canada/plugin/internal_plugin.py:82: in patched_upsert_data
    raise e
.../canada/plugin/internal_plugin.py:69: in patched_upsert_data
    return original_upsert_data(context, data_dict)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

context = {'__auth_audit': [], '__auth_user_obj_checked': True, 'auth_user_obj': <User id=bb096f29-9a92-47a7-8e07-1241c6da4624 n...ve image_url=None plugin_extras=None>, 'connection': <sqlalchemy.engine.base.Connection object at 0x7f5432b26520>, ...}
data_dict = {'connection_url': 'postgresql://datastore_write:pass@postgres/datastore_test', 'records': [{'automated_decision_syste...channel': ['EML', 'FAX', 'PERSON', 'ONL', 'POST', 'TEL'], ...}], 'resource_id': '24b243c8-709d-48ef-9a0f-7dcab86f3a34'}

    def upsert_data(context: Context, data_dict: dict[str, Any]):
        '''insert all data from records'''
        if not data_dict.get('records'):
            return
    
        method = data_dict.get('method', _UPSERT)
    
        fields = _get_fields(context['connection'], data_dict['resource_id'])
        field_names = _pluck('id', fields)
        records = data_dict['records']
        sql_columns = ", ".join(
            identifier(name) for name in field_names)
        num = -1
    
        if method == _INSERT:
            rows = []
            for num, record in enumerate(records):
                _validate_record(record, num, field_names)
    
                row = {}
                for idx, field in enumerate(fields):
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        value = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        value = None
                    row[f"val_{idx}"] = value
                rows.append(row)
    
            sql_string = '''INSERT INTO {res_id} ({columns})
                VALUES ({values});'''.format(
                res_id=identifier(data_dict['resource_id']),
                columns=sql_columns,
                values=', '.join([
                    f":val_{idx}" for idx in range(0, len(field_names))
                ])
            )
    
            try:
                context['connection'].execute(sa.text(sql_string), rows)
            except (DatabaseError, DataError) as err:
                # (canada fork only): parse constraint sql errors
                # TODO: upstream contrib!!
                errmsg = _programming_error_summary(err)
                if 'violates foreign key constraint' in errmsg:
                    _ = lambda x:x
                    errmsg = _(
                        'Cannot insert records ({refValues}) because '\
                        'they do not exist in the referenced table. '\
                        'Referencing {refKeys} from {refTable}.')
                    raise ValidationError(dict(
                        _parse_constraint_error_from_psql_error(err, errmsg),
                        records_row=num))
                raise ValidationError({
                    'records': [errmsg],
                    'records_row': num,
                })
    
        elif method in [_UPDATE, _UPSERT]:
            unique_keys = _get_unique_key(context, data_dict)
    
            for num, record in enumerate(records):
                if not unique_keys and '_id' not in record:
                    raise ValidationError({
                        'table': [u'unique key must be passed for update/upsert']
                    })
    
                elif '_id' not in record:
                    # all key columns have to be defined
                    missing_fields = [field for field in unique_keys
                                      if field not in record]
                    if missing_fields:
                        raise ValidationError({
                            'key': [u'''fields "{fields}" are missing
                                but needed as key'''.format(
                                    fields=', '.join(missing_fields))]
                        })
    
                for field in fields:
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        record[field['id']] = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        record[field['id']] = None
    
                non_existing_field_names = [
                    field for field in record
                    if field not in field_names and field != '_id'
                ]
                if non_existing_field_names:
>                   raise ValidationError({
                        'fields': [u'fields "{0}" do not exist'.format(
                            ', '.join(non_existing_field_names))]
                    })
E                   ckan.logic.ValidationError: None - {'fields': ['fields "service_name_en, service_name_fr" do not exist']}

.../datastore/backend/postgres.py:1577: ValidationError

ckanext/canada/tests/test_service.py::TestStdService::test_primary_key_commas

Stack Traces | 0.589s run time

self = <ckanext.canada.tests.test_service.TestStdService object at 0x7f7f322908e0>

    def test_primary_key_commas(self):
        """
        Commas in primary keys should error
        """
        record = get_chromo('service-std')['examples']['record'].copy().copy()
        record['service_id'] = 'this,is,a,failure'
        record['service_standard_id'] = 'this,is,a,failure'
        with pytest.raises(ValidationError) as ve:
            self.lc.action.datastore_upsert(
                resource_id=self.resource_id,
                records=[record])
        model.Session.rollback()
        err = ve.value.error_dict
>       assert 'records' in err
E       assert 'records' in {'fields': ['fields "service_name_en, service_name_fr" do not exist']}

.../canada/tests/test_service.py:397: AssertionError

ckanext/canada/tests/test_service.py::TestStdService::test_duolang_fields

Stack Traces | 0.598s run time

self = <ckanext.canada.tests.test_service.TestStdService object at 0x7f98aee31ca0>

    def test_duolang_fields(self):
        """
        Fields with both _en and _fr should require eachother
        """
>       self._make_parent_record()

.../canada/tests/test_service.py:465: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../canada/tests/test_service.py:370: in _make_parent_record
    self.lc.action.datastore_upsert(
.../ckanapi/ckanapi/common.py:60: in action
    return self._ckan.call_action(name, data_dict=kwargs)
.../ckanapi/ckanapi/localckan.py:74: in call_action
    return self._get_action(action)(context, data_dict)
.../ckan/logic/__init__.py:581: in wrapped
    result = _action(context, data_dict, **kw)
.../ckanext/recombinant/logic.py:517: in recombinant_datastore_upsert
    return up_func(context, data_dict)
.../datastore/logic/action.py:296: in datastore_upsert
    result = backend.upsert(context, data_dict)
.../datastore/backend/postgres.py:2575: in upsert
    return upsert(context, data_dict)
.../datastore/backend/postgres.py:2061: in upsert
    upsert_data(dict(context, connection=connection), data_dict)
.../canada/plugin/internal_plugin.py:82: in patched_upsert_data
    raise e
.../canada/plugin/internal_plugin.py:69: in patched_upsert_data
    return original_upsert_data(context, data_dict)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

context = {'__auth_audit': [], '__auth_user_obj_checked': True, 'auth_user_obj': <User id=dad1de07-9a9a-4ff8-8e2d-00dbe919e017 n...ve image_url=None plugin_extras=None>, 'connection': <sqlalchemy.engine.base.Connection object at 0x7f98af9feb80>, ...}
data_dict = {'connection_url': 'postgresql://datastore_write:pass@postgres/datastore_test', 'records': [{'automated_decision_syste...channel': ['EML', 'FAX', 'PERSON', 'ONL', 'POST', 'TEL'], ...}], 'resource_id': 'fbdfe299-8ac3-4742-aea7-a193e96a88d7'}

    def upsert_data(context: Context, data_dict: dict[str, Any]):
        '''insert all data from records'''
        if not data_dict.get('records'):
            return
    
        method = data_dict.get('method', _UPSERT)
    
        fields = _get_fields(context['connection'], data_dict['resource_id'])
        field_names = _pluck('id', fields)
        records = data_dict['records']
        sql_columns = ", ".join(
            identifier(name) for name in field_names)
        num = -1
    
        if method == _INSERT:
            rows = []
            for num, record in enumerate(records):
                _validate_record(record, num, field_names)
    
                row = {}
                for idx, field in enumerate(fields):
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        value = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        value = None
                    row[f"val_{idx}"] = value
                rows.append(row)
    
            sql_string = '''INSERT INTO {res_id} ({columns})
                VALUES ({values});'''.format(
                res_id=identifier(data_dict['resource_id']),
                columns=sql_columns,
                values=', '.join([
                    f":val_{idx}" for idx in range(0, len(field_names))
                ])
            )
    
            try:
                context['connection'].execute(sa.text(sql_string), rows)
            except (DatabaseError, DataError) as err:
                # (canada fork only): parse constraint sql errors
                # TODO: upstream contrib!!
                errmsg = _programming_error_summary(err)
                if 'violates foreign key constraint' in errmsg:
                    _ = lambda x:x
                    errmsg = _(
                        'Cannot insert records ({refValues}) because '\
                        'they do not exist in the referenced table. '\
                        'Referencing {refKeys} from {refTable}.')
                    raise ValidationError(dict(
                        _parse_constraint_error_from_psql_error(err, errmsg),
                        records_row=num))
                raise ValidationError({
                    'records': [errmsg],
                    'records_row': num,
                })
    
        elif method in [_UPDATE, _UPSERT]:
            unique_keys = _get_unique_key(context, data_dict)
    
            for num, record in enumerate(records):
                if not unique_keys and '_id' not in record:
                    raise ValidationError({
                        'table': [u'unique key must be passed for update/upsert']
                    })
    
                elif '_id' not in record:
                    # all key columns have to be defined
                    missing_fields = [field for field in unique_keys
                                      if field not in record]
                    if missing_fields:
                        raise ValidationError({
                            'key': [u'''fields "{fields}" are missing
                                but needed as key'''.format(
                                    fields=', '.join(missing_fields))]
                        })
    
                for field in fields:
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        record[field['id']] = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        record[field['id']] = None
    
                non_existing_field_names = [
                    field for field in record
                    if field not in field_names and field != '_id'
                ]
                if non_existing_field_names:
>                   raise ValidationError({
                        'fields': [u'fields "{0}" do not exist'.format(
                            ', '.join(non_existing_field_names))]
                    })
E                   ckan.logic.ValidationError: None - {'fields': ['fields "service_name_en, service_name_fr" do not exist']}

.../datastore/backend/postgres.py:1577: ValidationError

ckanext/canada/tests/test_service.py::TestService::test_conditional_fields

Stack Traces | 0.622s run time

self = <ckanext.canada.tests.test_service.TestService object at 0x7f7807784b50>

    def test_conditional_fields(self):
        """
        Test conditionally required fields
        """
        chromo = get_chromo('service')
        record = chromo['examples']['record'].copy()
    
        # automated_decision_system_description_en and automated_decision_system_description_fr
        # should be required if automated_decision_system is Y.
        record['automated_decision_system'] = 'Y'
        with pytest.raises(ValidationError) as ve:
            self.lc.action.datastore_upsert(
                resource_id=self.resource_id,
                records=[record])
        model.Session.rollback()
        err = ve.value.error_dict
>       assert 'records' in err
E       assert 'records' in {'fields': ['fields "service_name_en, service_name_fr" do not exist']}

.../canada/tests/test_service.py:144: AssertionError

ckanext/canada/tests/test_service.py::TestService::test_foreign_constraint

Stack Traces | 0.623s run time

self = <ckanext.canada.tests.test_service.TestService object at 0x7f98afd91550>

    def test_foreign_constraint(self):
        """
        Trying to delete a Service record when there are Standard records referencing it should raise an exception
        """
        record = get_chromo('service')['examples']['record'].copy()
>       self.lc.action.datastore_upsert(
            resource_id=self.resource_id,
            records=[record])

.../canada/tests/test_service.py:60: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../ckanapi/ckanapi/common.py:60: in action
    return self._ckan.call_action(name, data_dict=kwargs)
.../ckanapi/ckanapi/localckan.py:74: in call_action
    return self._get_action(action)(context, data_dict)
.../ckan/logic/__init__.py:581: in wrapped
    result = _action(context, data_dict, **kw)
.../ckanext/recombinant/logic.py:517: in recombinant_datastore_upsert
    return up_func(context, data_dict)
.../datastore/logic/action.py:296: in datastore_upsert
    result = backend.upsert(context, data_dict)
.../datastore/backend/postgres.py:2575: in upsert
    return upsert(context, data_dict)
.../datastore/backend/postgres.py:2061: in upsert
    upsert_data(dict(context, connection=connection), data_dict)
.../canada/plugin/internal_plugin.py:82: in patched_upsert_data
    raise e
.../canada/plugin/internal_plugin.py:69: in patched_upsert_data
    return original_upsert_data(context, data_dict)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

context = {'__auth_audit': [], '__auth_user_obj_checked': True, 'auth_user_obj': <User id=dad1de07-9a9a-4ff8-8e2d-00dbe919e017 n...ve image_url=None plugin_extras=None>, 'connection': <sqlalchemy.engine.base.Connection object at 0x7f98af0ebd60>, ...}
data_dict = {'connection_url': 'postgresql://datastore_write:pass@postgres/datastore_test', 'records': [{'automated_decision_syste...channel': ['EML', 'FAX', 'PERSON', 'ONL', 'POST', 'TEL'], ...}], 'resource_id': '97e47816-6205-4219-a92c-c74f4edd4b1f'}

    def upsert_data(context: Context, data_dict: dict[str, Any]):
        '''insert all data from records'''
        if not data_dict.get('records'):
            return
    
        method = data_dict.get('method', _UPSERT)
    
        fields = _get_fields(context['connection'], data_dict['resource_id'])
        field_names = _pluck('id', fields)
        records = data_dict['records']
        sql_columns = ", ".join(
            identifier(name) for name in field_names)
        num = -1
    
        if method == _INSERT:
            rows = []
            for num, record in enumerate(records):
                _validate_record(record, num, field_names)
    
                row = {}
                for idx, field in enumerate(fields):
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        value = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        value = None
                    row[f"val_{idx}"] = value
                rows.append(row)
    
            sql_string = '''INSERT INTO {res_id} ({columns})
                VALUES ({values});'''.format(
                res_id=identifier(data_dict['resource_id']),
                columns=sql_columns,
                values=', '.join([
                    f":val_{idx}" for idx in range(0, len(field_names))
                ])
            )
    
            try:
                context['connection'].execute(sa.text(sql_string), rows)
            except (DatabaseError, DataError) as err:
                # (canada fork only): parse constraint sql errors
                # TODO: upstream contrib!!
                errmsg = _programming_error_summary(err)
                if 'violates foreign key constraint' in errmsg:
                    _ = lambda x:x
                    errmsg = _(
                        'Cannot insert records ({refValues}) because '\
                        'they do not exist in the referenced table. '\
                        'Referencing {refKeys} from {refTable}.')
                    raise ValidationError(dict(
                        _parse_constraint_error_from_psql_error(err, errmsg),
                        records_row=num))
                raise ValidationError({
                    'records': [errmsg],
                    'records_row': num,
                })
    
        elif method in [_UPDATE, _UPSERT]:
            unique_keys = _get_unique_key(context, data_dict)
    
            for num, record in enumerate(records):
                if not unique_keys and '_id' not in record:
                    raise ValidationError({
                        'table': [u'unique key must be passed for update/upsert']
                    })
    
                elif '_id' not in record:
                    # all key columns have to be defined
                    missing_fields = [field for field in unique_keys
                                      if field not in record]
                    if missing_fields:
                        raise ValidationError({
                            'key': [u'''fields "{fields}" are missing
                                but needed as key'''.format(
                                    fields=', '.join(missing_fields))]
                        })
    
                for field in fields:
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        record[field['id']] = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        record[field['id']] = None
    
                non_existing_field_names = [
                    field for field in record
                    if field not in field_names and field != '_id'
                ]
                if non_existing_field_names:
>                   raise ValidationError({
                        'fields': [u'fields "{0}" do not exist'.format(
                            ', '.join(non_existing_field_names))]
                    })
E                   ckan.logic.ValidationError: None - {'fields': ['fields "service_name_en, service_name_fr" do not exist']}

.../datastore/backend/postgres.py:1577: ValidationError

ckanext/canada/tests/test_service.py::TestService::test_required_fields

Stack Traces | 0.626s run time

self = <ckanext.canada.tests.test_service.TestService object at 0x7f7f32f341f0>

    def test_required_fields(self):
        """
        Excluding required fields should raise an exception
        """
        chromo = get_chromo('service')
        record = chromo['examples']['record'].copy()
    
        expected_required_fields = ['service_name_en', 'service_name_fr',
                                    'service_description_en', 'service_description_fr',
                                    'service_type', 'service_recipient_type',
                                    'service_scope', 'client_target_groups',
                                    'program_id', 'client_feedback_channel',
                                    'automated_decision_system', 'service_fee',
                                    'os_account_registration', 'os_authentication',
                                    'os_application', 'os_decision', 'os_issuance',
                                    'os_issue_resolution_feedback', 'sin_usage',
                                    'cra_bn_identifier_usage', 'num_phone_enquiries',
                                    'num_applications_by_phone', 'num_website_visits',
                                    'num_applications_online', 'num_applications_in_person',
                                    'num_applications_by_mail', 'num_applications_by_email',
                                    'num_applications_by_fax', 'num_applications_by_other']
    
        for field in chromo['fields']:
            if field['datastore_id'] in chromo['datastore_primary_key']:
                continue
            if field.get('excel_required') or field.get('form_required'):
                assert field['datastore_id'] in expected_required_fields
                record[field['datastore_id']] = None
    
        with pytest.raises(ValidationError) as ve:
            self.lc.action.datastore_upsert(
                resource_id=self.resource_id,
                records=[record])
        model.Session.rollback()
        err = ve.value.error_dict
>       assert 'records' in err
E       assert 'records' in {'fields': ['fields "service_name_en, service_name_fr" do not exist']}

.../canada/tests/test_service.py:124: AssertionError

ckanext/canada/tests/test_service.py::TestService::test_choice_fields

Stack Traces | 0.648s run time

self = <ckanext.canada.tests.test_service.TestService object at 0x7f59074e6790>

    def test_choice_fields(self):
        """
        Fields with choices should expect those values
        """
        chromo = get_chromo('service')
        record = chromo['examples']['record'].copy()
    
        expected_choice_fields = ['fiscal_yr', 'service_type', 'service_recipient_type',
                                  'service_scope', 'client_target_groups', 'program_id',
                                  'client_feedback_channel', 'automated_decision_system',
                                  'service_fee', 'os_account_registration',
                                  'os_authentication', 'os_application', 'os_decision',
                                  'os_issuance', 'os_issue_resolution_feedback',
                                  'last_service_review', 'last_service_improvement',
                                  'sin_usage', 'cra_bn_identifier_usage']
    
        for field in chromo['fields']:
            if field.get('published_resource_computed_field'):
                continue
            if 'choices_file' in field or 'choices' in field:
                assert field['datastore_id'] in expected_choice_fields
                if field['datastore_type'] == '_text':
                    record[field['datastore_id']] = ['zzz']
                else:
                    record[field['datastore_id']] = 'zzz'
    
        with pytest.raises(ValidationError) as ve:
            self.lc.action.datastore_upsert(
                resource_id=self.resource_id,
                records=[record])
        model.Session.rollback()
        err = ve.value.error_dict
>       assert 'records' in err
E       assert 'records' in {'fields': ['fields "service_name_en, service_name_fr" do not exist']}

.../canada/tests/test_service.py:275: AssertionError

ckanext/canada/tests/test_service.py::TestStdService::test_example

Stack Traces | 0.662s run time

self = <ckanext.canada.tests.test_service.TestStdService object at 0x7f5907845790>

    def test_example(self):
        """
        Example data should load
        """
>       self._make_parent_record()

.../canada/tests/test_service.py:378: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../canada/tests/test_service.py:370: in _make_parent_record
    self.lc.action.datastore_upsert(
.../ckanapi/ckanapi/common.py:60: in action
    return self._ckan.call_action(name, data_dict=kwargs)
.../ckanapi/ckanapi/localckan.py:74: in call_action
    return self._get_action(action)(context, data_dict)
.../ckan/logic/__init__.py:581: in wrapped
    result = _action(context, data_dict, **kw)
.../ckanext/recombinant/logic.py:517: in recombinant_datastore_upsert
    return up_func(context, data_dict)
.../datastore/logic/action.py:296: in datastore_upsert
    result = backend.upsert(context, data_dict)
.../datastore/backend/postgres.py:2575: in upsert
    return upsert(context, data_dict)
.../datastore/backend/postgres.py:2061: in upsert
    upsert_data(dict(context, connection=connection), data_dict)
.../canada/plugin/internal_plugin.py:82: in patched_upsert_data
    raise e
.../canada/plugin/internal_plugin.py:69: in patched_upsert_data
    return original_upsert_data(context, data_dict)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

context = {'__auth_audit': [], '__auth_user_obj_checked': True, 'auth_user_obj': <User id=4cce2f20-4ba0-40be-a005-8b5410f1c86f n...ve image_url=None plugin_extras=None>, 'connection': <sqlalchemy.engine.base.Connection object at 0x7f590b5b06d0>, ...}
data_dict = {'connection_url': 'postgresql://datastore_write:pass@postgres/datastore_test', 'records': [{'automated_decision_syste...channel': ['EML', 'FAX', 'PERSON', 'ONL', 'POST', 'TEL'], ...}], 'resource_id': '7d584525-f7bb-438e-b2c9-dd6c635f7c98'}

    def upsert_data(context: Context, data_dict: dict[str, Any]):
        '''insert all data from records'''
        if not data_dict.get('records'):
            return
    
        method = data_dict.get('method', _UPSERT)
    
        fields = _get_fields(context['connection'], data_dict['resource_id'])
        field_names = _pluck('id', fields)
        records = data_dict['records']
        sql_columns = ", ".join(
            identifier(name) for name in field_names)
        num = -1
    
        if method == _INSERT:
            rows = []
            for num, record in enumerate(records):
                _validate_record(record, num, field_names)
    
                row = {}
                for idx, field in enumerate(fields):
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        value = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        value = None
                    row[f"val_{idx}"] = value
                rows.append(row)
    
            sql_string = '''INSERT INTO {res_id} ({columns})
                VALUES ({values});'''.format(
                res_id=identifier(data_dict['resource_id']),
                columns=sql_columns,
                values=', '.join([
                    f":val_{idx}" for idx in range(0, len(field_names))
                ])
            )
    
            try:
                context['connection'].execute(sa.text(sql_string), rows)
            except (DatabaseError, DataError) as err:
                # (canada fork only): parse constraint sql errors
                # TODO: upstream contrib!!
                errmsg = _programming_error_summary(err)
                if 'violates foreign key constraint' in errmsg:
                    _ = lambda x:x
                    errmsg = _(
                        'Cannot insert records ({refValues}) because '\
                        'they do not exist in the referenced table. '\
                        'Referencing {refKeys} from {refTable}.')
                    raise ValidationError(dict(
                        _parse_constraint_error_from_psql_error(err, errmsg),
                        records_row=num))
                raise ValidationError({
                    'records': [errmsg],
                    'records_row': num,
                })
    
        elif method in [_UPDATE, _UPSERT]:
            unique_keys = _get_unique_key(context, data_dict)
    
            for num, record in enumerate(records):
                if not unique_keys and '_id' not in record:
                    raise ValidationError({
                        'table': [u'unique key must be passed for update/upsert']
                    })
    
                elif '_id' not in record:
                    # all key columns have to be defined
                    missing_fields = [field for field in unique_keys
                                      if field not in record]
                    if missing_fields:
                        raise ValidationError({
                            'key': [u'''fields "{fields}" are missing
                                but needed as key'''.format(
                                    fields=', '.join(missing_fields))]
                        })
    
                for field in fields:
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        record[field['id']] = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        record[field['id']] = None
    
                non_existing_field_names = [
                    field for field in record
                    if field not in field_names and field != '_id'
                ]
                if non_existing_field_names:
>                   raise ValidationError({
                        'fields': [u'fields "{0}" do not exist'.format(
                            ', '.join(non_existing_field_names))]
                    })
E                   ckan.logic.ValidationError: None - {'fields': ['fields "service_name_en, service_name_fr" do not exist']}

.../datastore/backend/postgres.py:1577: ValidationError

ckanext/canada/tests/test_service.py::TestStdService::test_filter_script

Stack Traces | 0.696s run time

self = <ckanext.canada.tests.test_service.TestStdService object at 0x7fdcb34f49d0>

    def test_filter_script(self):
        """
        The calculations for performance and target_met
        should be correct for the required Business Logic
    
        NOTE: csv.DictReader treats every dict value as a string,
              so we need to use Strings here for target, volume_meeting_target,
              and total_volume. Empty strings ("") is None.
    
        NOTE: the filter test returns a Dict, not a csv.DictWriter,
              so we can assert on object types here.
        """
>       self._make_parent_record()

.../canada/tests/test_service.py:629: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../canada/tests/test_service.py:370: in _make_parent_record
    self.lc.action.datastore_upsert(
.../ckanapi/ckanapi/common.py:60: in action
    return self._ckan.call_action(name, data_dict=kwargs)
.../ckanapi/ckanapi/localckan.py:74: in call_action
    return self._get_action(action)(context, data_dict)
.../ckan/logic/__init__.py:581: in wrapped
    result = _action(context, data_dict, **kw)
.../ckanext/recombinant/logic.py:517: in recombinant_datastore_upsert
    return up_func(context, data_dict)
.../datastore/logic/action.py:296: in datastore_upsert
    result = backend.upsert(context, data_dict)
.../datastore/backend/postgres.py:2575: in upsert
    return upsert(context, data_dict)
.../datastore/backend/postgres.py:2061: in upsert
    upsert_data(dict(context, connection=connection), data_dict)
.../canada/plugin/internal_plugin.py:82: in patched_upsert_data
    raise e
.../canada/plugin/internal_plugin.py:69: in patched_upsert_data
    return original_upsert_data(context, data_dict)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

context = {'__auth_audit': [], '__auth_user_obj_checked': True, 'auth_user_obj': <User id=02433257-8a61-4794-bb47-d13718b1b663 n...ve image_url=None plugin_extras=None>, 'connection': <sqlalchemy.engine.base.Connection object at 0x7fdcb2394340>, ...}
data_dict = {'connection_url': 'postgresql://datastore_write:pass@postgres/datastore_test', 'records': [{'automated_decision_syste...channel': ['EML', 'FAX', 'PERSON', 'ONL', 'POST', 'TEL'], ...}], 'resource_id': 'e9217a71-7015-4d83-8067-32198d388f79'}

    def upsert_data(context: Context, data_dict: dict[str, Any]):
        '''insert all data from records'''
        if not data_dict.get('records'):
            return
    
        method = data_dict.get('method', _UPSERT)
    
        fields = _get_fields(context['connection'], data_dict['resource_id'])
        field_names = _pluck('id', fields)
        records = data_dict['records']
        sql_columns = ", ".join(
            identifier(name) for name in field_names)
        num = -1
    
        if method == _INSERT:
            rows = []
            for num, record in enumerate(records):
                _validate_record(record, num, field_names)
    
                row = {}
                for idx, field in enumerate(fields):
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        value = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        value = None
                    row[f"val_{idx}"] = value
                rows.append(row)
    
            sql_string = '''INSERT INTO {res_id} ({columns})
                VALUES ({values});'''.format(
                res_id=identifier(data_dict['resource_id']),
                columns=sql_columns,
                values=', '.join([
                    f":val_{idx}" for idx in range(0, len(field_names))
                ])
            )
    
            try:
                context['connection'].execute(sa.text(sql_string), rows)
            except (DatabaseError, DataError) as err:
                # (canada fork only): parse constraint sql errors
                # TODO: upstream contrib!!
                errmsg = _programming_error_summary(err)
                if 'violates foreign key constraint' in errmsg:
                    _ = lambda x:x
                    errmsg = _(
                        'Cannot insert records ({refValues}) because '\
                        'they do not exist in the referenced table. '\
                        'Referencing {refKeys} from {refTable}.')
                    raise ValidationError(dict(
                        _parse_constraint_error_from_psql_error(err, errmsg),
                        records_row=num))
                raise ValidationError({
                    'records': [errmsg],
                    'records_row': num,
                })
    
        elif method in [_UPDATE, _UPSERT]:
            unique_keys = _get_unique_key(context, data_dict)
    
            for num, record in enumerate(records):
                if not unique_keys and '_id' not in record:
                    raise ValidationError({
                        'table': [u'unique key must be passed for update/upsert']
                    })
    
                elif '_id' not in record:
                    # all key columns have to be defined
                    missing_fields = [field for field in unique_keys
                                      if field not in record]
                    if missing_fields:
                        raise ValidationError({
                            'key': [u'''fields "{fields}" are missing
                                but needed as key'''.format(
                                    fields=', '.join(missing_fields))]
                        })
    
                for field in fields:
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        record[field['id']] = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        record[field['id']] = None
    
                non_existing_field_names = [
                    field for field in record
                    if field not in field_names and field != '_id'
                ]
                if non_existing_field_names:
>                   raise ValidationError({
                        'fields': [u'fields "{0}" do not exist'.format(
                            ', '.join(non_existing_field_names))]
                    })
E                   ckan.logic.ValidationError: None - {'fields': ['fields "service_name_en, service_name_fr" do not exist']}

.../datastore/backend/postgres.py:1577: ValidationError

ckanext/canada/tests/test_service.py::TestService::test_example

Stack Traces | 0.828s run time

self = <ckanext.canada.tests.test_service.TestService object at 0x7fb7e0173fd0>

    def test_example(self):
        """
        Example data should load
        """
        record = get_chromo('service')['examples']['record'].copy()
>       self.lc.action.datastore_upsert(
            resource_id=self.resource_id,
            records=[record])

.../canada/tests/test_service.py:35: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../ckanapi/ckanapi/common.py:60: in action
    return self._ckan.call_action(name, data_dict=kwargs)
.../ckanapi/ckanapi/localckan.py:74: in call_action
    return self._get_action(action)(context, data_dict)
.../ckan/logic/__init__.py:581: in wrapped
    result = _action(context, data_dict, **kw)
.../ckanext/recombinant/logic.py:517: in recombinant_datastore_upsert
    return up_func(context, data_dict)
.../datastore/logic/action.py:296: in datastore_upsert
    result = backend.upsert(context, data_dict)
.../datastore/backend/postgres.py:2575: in upsert
    return upsert(context, data_dict)
.../datastore/backend/postgres.py:2061: in upsert
    upsert_data(dict(context, connection=connection), data_dict)
.../canada/plugin/internal_plugin.py:82: in patched_upsert_data
    raise e
.../canada/plugin/internal_plugin.py:69: in patched_upsert_data
    return original_upsert_data(context, data_dict)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

context = {'__auth_audit': [], '__auth_user_obj_checked': True, 'auth_user_obj': <User id=68edb0fa-0206-41f7-8919-8d80ce62b844 n...ve image_url=None plugin_extras=None>, 'connection': <sqlalchemy.engine.base.Connection object at 0x7fb7dcd0a4c0>, ...}
data_dict = {'connection_url': 'postgresql://datastore_write:pass@postgres/datastore_test', 'records': [{'automated_decision_syste...channel': ['EML', 'FAX', 'PERSON', 'ONL', 'POST', 'TEL'], ...}], 'resource_id': '6d0d4942-1a0b-4f1f-abfc-02cf8137e0aa'}

    def upsert_data(context: Context, data_dict: dict[str, Any]):
        '''insert all data from records'''
        if not data_dict.get('records'):
            return
    
        method = data_dict.get('method', _UPSERT)
    
        fields = _get_fields(context['connection'], data_dict['resource_id'])
        field_names = _pluck('id', fields)
        records = data_dict['records']
        sql_columns = ", ".join(
            identifier(name) for name in field_names)
        num = -1
    
        if method == _INSERT:
            rows = []
            for num, record in enumerate(records):
                _validate_record(record, num, field_names)
    
                row = {}
                for idx, field in enumerate(fields):
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        value = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        value = None
                    row[f"val_{idx}"] = value
                rows.append(row)
    
            sql_string = '''INSERT INTO {res_id} ({columns})
                VALUES ({values});'''.format(
                res_id=identifier(data_dict['resource_id']),
                columns=sql_columns,
                values=', '.join([
                    f":val_{idx}" for idx in range(0, len(field_names))
                ])
            )
    
            try:
                context['connection'].execute(sa.text(sql_string), rows)
            except (DatabaseError, DataError) as err:
                # (canada fork only): parse constraint sql errors
                # TODO: upstream contrib!!
                errmsg = _programming_error_summary(err)
                if 'violates foreign key constraint' in errmsg:
                    _ = lambda x:x
                    errmsg = _(
                        'Cannot insert records ({refValues}) because '\
                        'they do not exist in the referenced table. '\
                        'Referencing {refKeys} from {refTable}.')
                    raise ValidationError(dict(
                        _parse_constraint_error_from_psql_error(err, errmsg),
                        records_row=num))
                raise ValidationError({
                    'records': [errmsg],
                    'records_row': num,
                })
    
        elif method in [_UPDATE, _UPSERT]:
            unique_keys = _get_unique_key(context, data_dict)
    
            for num, record in enumerate(records):
                if not unique_keys and '_id' not in record:
                    raise ValidationError({
                        'table': [u'unique key must be passed for update/upsert']
                    })
    
                elif '_id' not in record:
                    # all key columns have to be defined
                    missing_fields = [field for field in unique_keys
                                      if field not in record]
                    if missing_fields:
                        raise ValidationError({
                            'key': [u'''fields "{fields}" are missing
                                but needed as key'''.format(
                                    fields=', '.join(missing_fields))]
                        })
    
                for field in fields:
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        record[field['id']] = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        record[field['id']] = None
    
                non_existing_field_names = [
                    field for field in record
                    if field not in field_names and field != '_id'
                ]
                if non_existing_field_names:
>                   raise ValidationError({
                        'fields': [u'fields "{0}" do not exist'.format(
                            ', '.join(non_existing_field_names))]
                    })
E                   ckan.logic.ValidationError: None - {'fields': ['fields "service_name_en, service_name_fr" do not exist']}

.../datastore/backend/postgres.py:1577: ValidationError

ckanext/canada/tests/test_service.py::TestService::test_duolang_fields

Stack Traces | 0.878s run time

self = <ckanext.canada.tests.test_service.TestService object at 0x7fdcb4c91e80>

    def test_duolang_fields(self):
        """
        Fields with both _en and _fr should require eachother
        """
        chromo = get_chromo('service')
        record = chromo['examples']['record'].copy()
    
        record['num_applications_by_other'] = 0
    
        record['automated_decision_system_description_en'] = 'Not Blank'
        record['automated_decision_system_description_fr'] = None
    
        record['os_comments_client_interaction_en'] = 'Not Blank'
        record['os_comments_client_interaction_fr'] = None
    
        record['special_remarks_en'] = 'Not Blank'
        record['special_remarks_fr'] = None
    
        record['service_uri_en'] = 'Not Blank'
        record['service_uri_fr'] = None
    
        with pytest.raises(ValidationError) as ve:
            self.lc.action.datastore_upsert(
                resource_id=self.resource_id,
                records=[record])
        model.Session.rollback()
        err = ve.value.error_dict
>       assert 'records' in err
E       assert 'records' in {'fields': ['fields "service_name_en, service_name_fr" do not exist']}

.../canada/tests/test_service.py:213: AssertionError

View the full list of 1 ❄️ flaky test(s)

ckanext/canada/tests/test_make.py::TestMakePD::test_make_service

Flake rate in main: 14.29% (Passed 6 times, Failed 1 times)

Stack Traces | 30.5s run time

self = <ckanext.canada.tests.test_make.TestMakePD object at 0x7f590baa6940>

    def test_make_service(self):
        assert self.ckan_ini
        self._setup_ini(self.ckan_ini)
    
>       self._setup_pd(type='service', nil_type='service-std')

.../canada/tests/test_make.py:534: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../canada/tests/test_make.py:129: in _setup_pd
    self.action.datastore_upsert(
.../ckanapi/ckanapi/common.py:60: in action
    return self._ckan.call_action(name, data_dict=kwargs)
.../ckanapi/ckanapi/localckan.py:74: in call_action
    return self._get_action(action)(context, data_dict)
.../ckan/logic/__init__.py:581: in wrapped
    result = _action(context, data_dict, **kw)
.../ckanext/recombinant/logic.py:517: in recombinant_datastore_upsert
    return up_func(context, data_dict)
.../datastore/logic/action.py:296: in datastore_upsert
    result = backend.upsert(context, data_dict)
.../datastore/backend/postgres.py:2575: in upsert
    return upsert(context, data_dict)
.../datastore/backend/postgres.py:2061: in upsert
    upsert_data(dict(context, connection=connection), data_dict)
.../canada/plugin/internal_plugin.py:82: in patched_upsert_data
    raise e
.../canada/plugin/internal_plugin.py:69: in patched_upsert_data
    return original_upsert_data(context, data_dict)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

context = {'__auth_audit': [], '__auth_user_obj_checked': True, 'auth_user_obj': <User id=4cce2f20-4ba0-40be-a005-8b5410f1c86f n...ve image_url=None plugin_extras=None>, 'connection': <sqlalchemy.engine.base.Connection object at 0x7f5906deba60>, ...}
data_dict = {'connection_url': 'postgresql://datastore_write:pass@postgres/datastore_test', 'records': [{'automated_decision_syste...channel': ['EML', 'FAX', 'PERSON', 'ONL', 'POST', 'TEL'], ...}], 'resource_id': 'a92b6b7e-ef76-4143-ab0f-f8d4d0da6bbc'}

    def upsert_data(context: Context, data_dict: dict[str, Any]):
        '''insert all data from records'''
        if not data_dict.get('records'):
            return
    
        method = data_dict.get('method', _UPSERT)
    
        fields = _get_fields(context['connection'], data_dict['resource_id'])
        field_names = _pluck('id', fields)
        records = data_dict['records']
        sql_columns = ", ".join(
            identifier(name) for name in field_names)
        num = -1
    
        if method == _INSERT:
            rows = []
            for num, record in enumerate(records):
                _validate_record(record, num, field_names)
    
                row = {}
                for idx, field in enumerate(fields):
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        value = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        value = None
                    row[f"val_{idx}"] = value
                rows.append(row)
    
            sql_string = '''INSERT INTO {res_id} ({columns})
                VALUES ({values});'''.format(
                res_id=identifier(data_dict['resource_id']),
                columns=sql_columns,
                values=', '.join([
                    f":val_{idx}" for idx in range(0, len(field_names))
                ])
            )
    
            try:
                context['connection'].execute(sa.text(sql_string), rows)
            except (DatabaseError, DataError) as err:
                # (canada fork only): parse constraint sql errors
                # TODO: upstream contrib!!
                errmsg = _programming_error_summary(err)
                if 'violates foreign key constraint' in errmsg:
                    _ = lambda x:x
                    errmsg = _(
                        'Cannot insert records ({refValues}) because '\
                        'they do not exist in the referenced table. '\
                        'Referencing {refKeys} from {refTable}.')
                    raise ValidationError(dict(
                        _parse_constraint_error_from_psql_error(err, errmsg),
                        records_row=num))
                raise ValidationError({
                    'records': [errmsg],
                    'records_row': num,
                })
    
        elif method in [_UPDATE, _UPSERT]:
            unique_keys = _get_unique_key(context, data_dict)
    
            for num, record in enumerate(records):
                if not unique_keys and '_id' not in record:
                    raise ValidationError({
                        'table': [u'unique key must be passed for update/upsert']
                    })
    
                elif '_id' not in record:
                    # all key columns have to be defined
                    missing_fields = [field for field in unique_keys
                                      if field not in record]
                    if missing_fields:
                        raise ValidationError({
                            'key': [u'''fields "{fields}" are missing
                                but needed as key'''.format(
                                    fields=', '.join(missing_fields))]
                        })
    
                for field in fields:
                    value = record.get(field['id'])
                    if value is not None and field['type'].lower() == 'nested':
                        # a tuple with an empty second value
                        record[field['id']] = (json.dumps(value), '')
                    elif value == '' and field['type'] != 'text':
                        record[field['id']] = None
    
                non_existing_field_names = [
                    field for field in record
                    if field not in field_names and field != '_id'
                ]
                if non_existing_field_names:
>                   raise ValidationError({
                        'fields': [u'fields "{0}" do not exist'.format(
                            ', '.join(non_existing_field_names))]
                    })
E                   ckan.logic.ValidationError: None - {'fields': ['fields "service_name_en, service_name_fr" do not exist']}

.../datastore/backend/postgres.py:1577: ValidationError

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}

- Created new workflow to semi-automate stuffs.

- Removed service name computed fields.

- Trigger type cast.

JVickery-TBS · 2026-04-27T15:05:52Z

TODO: update any tests for test_service.py and its make tests if needed.

- Add db table and logic for reference data file hashes during load.

JVickery-TBS added 6 commits March 27, 2026 20:25

feat(dev): ref tables;

505d76e

- Started working on reference tables for pd data.

Merge branch 'master' into feature/ds-reference-tables

ec4635d

Merge branch 'master' into feature/ds-reference-tables

45fae29

feat(dev): ds ref tables cont;

2e72b9c

- Continued script to generate service reference data. - Continued schema for ref tables. - Improved pd subcommand for loading ref data.

feat(dev): ds ref tables cont;

10256b8

- Finalized script to generate service ref data. - Added more recombinant schema key/values.

feat(dev): ds ref tables cont;

fc740de

- Made JSONB index. - Added choices_filter_query. - Added recombinant org_name to the temp ds table. - Started writing the database validation.

JVickery-TBS requested a review from wardi April 9, 2026 20:13

JVickery-TBS assigned wardi and JVickery-TBS Apr 9, 2026

JVickery-TBS added enhancement Business Requirement labels Apr 9, 2026

JVickery-TBS mentioned this pull request Apr 9, 2026

Recombinant Org Based Triggers #1602

Closed

wardi reviewed Apr 10, 2026

View reviewed changes

Comment thread ckanext/canada/plugin/internal_plugin.py Outdated

JVickery-TBS added 3 commits April 16, 2026 19:40

feat(dev): ds ref tables cont;

433e216

- Finalized ref value checks and errors in the db func. - Fixed up some other logic things.

feat(dev): ds fiscal year func;

bcf369b

- Started working on fiscal year function.

feat(dev): ds fiscal year func;

65e40a8

- Finalized fiscal year db function.

wardi reviewed Apr 17, 2026

View reviewed changes

Comment thread bin/service_generate_reference_data.py Outdated

wardi reviewed Apr 17, 2026

View reviewed changes

Comment thread ckanext/canada/tables/service.yaml Outdated

wardi reviewed Apr 18, 2026

View reviewed changes

Comment thread ckanext/canada/tables/service.yaml

wardi reviewed Apr 18, 2026

View reviewed changes

Comment thread ckanext/canada/tables/service.yaml Outdated

JVickery-TBS added 5 commits April 20, 2026 15:23

feat(misc): changelog;

9af0578

- Added change log file.

fix(misc): syntax;

163fd73

- Pyright fixes. - Flake8 fixes.

feat(pd): service inventory;

eb89a03

- Made service name fields computed ones. - Modified filter scripts for service name lookups.

fix(pd): service inventory;

dfd01d8

- Filter script typo.

fix(pd): service inventory;

046e8f1

- Filter script typo.

JVickery-TBS added 3 commits April 21, 2026 13:32

feat(dev): github action;

1bd0da2

- Created new workflow to semi-automate stuffs.

fix(tests): service inventory;

6a9a1d1

- Removed service name computed fields.

fix(pd): service inventory;

9a82bb7

- Trigger type cast.

wardi reviewed Apr 24, 2026

View reviewed changes

Comment thread ckanext/canada/triggers.py Outdated

wardi reviewed Apr 24, 2026

View reviewed changes

Comment thread ckanext/canada/tables/service.yaml Outdated

wardi approved these changes Apr 24, 2026

View reviewed changes

fix(pd): service inventory;

70a1eab

- Trigger type cast.

feat(db): ref data hash;

4a2d1b3

- Add db table and logic for reference data file hashes during load.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Organization Validation for Service Inventory#1668

Organization Validation for Service Inventory#1668
JVickery-TBS wants to merge 19 commits into
masterfrom
feature/ds-reference-tables

JVickery-TBS commented Apr 9, 2026

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

wardi commented Apr 18, 2026

Uh oh!

codecov Bot commented Apr 20, 2026 •

edited

Loading

Uh oh!

Uh oh!

Uh oh!

JVickery-TBS commented Apr 27, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

JVickery-TBS commented Apr 9, 2026

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

wardi commented Apr 18, 2026

Uh oh!

codecov Bot commented Apr 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

❌ 19 Tests Failed:

Uh oh!

Uh oh!

Uh oh!

JVickery-TBS commented Apr 27, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

codecov Bot commented Apr 20, 2026 •

edited

Loading