Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use fix_ids method in localstack

Best Python code snippet using localstack_python

484e28958b27_deduplicate_packages.py

Source:484e28958b27_deduplicate_packages.py

1"""2deduplicate packages, issue 6173Revision ID: 484e28958b274Revises: 50e68db97d0a5Create Date: 2022-05-25 10:54:34.8326256"""7import logging8import sqlalchemy as sa9from alembic import op10revision = '484e28958b27'11down_revision = '50e68db97d0a'12log = logging.getLogger(__name__)13def upgrade():14    """15    Before we drop the Package.copr_dir_id argument, use it to "guess" the16    "main" package within each Copr.  That is to be kept, rest of the Package17    of the same name are to be removed.18    """19    session = sa.orm.sessionmaker(bind=op.get_bind())()20    # About 9k items here, 28k packages for de-duplication.21    duplications = session.execute(22        """23        select count(package.name), package.name, copr.id24        from package25        join copr on copr.id = package.copr_id26        group by copr.id, package.name27        having count(package.name) > 1;28        """29    )30    for duplication in duplications:31        _, package_name, copr_id = duplication32        fix_package_ids_result = session.execute(33            """34            select package.id35            from package36            left join copr_dir on package.copr_dir_id = copr_dir.id37            where package.copr_id = :copr_id and package.name = :package_name38            order by copr_dir.main = false asc, package.id asc;39            """,40            {"copr_id": copr_id, "package_name": package_name},41        )42        fix_ids = []43        main_package_id = None44        for row in fix_package_ids_result:45            package_id = row[0]46            if main_package_id is None:47                main_package_id = package_id48            else:49                fix_ids.append(package_id)50        log.debug(51            "Deduplicating %s=%s in copr_id=%s, affected packages: %s",52            package_name, main_package_id, copr_id,53            ", ".join([str(x) for x in fix_ids]),54        )55        session.execute(56            """57            update build set package_id = :package_id58            where package_id in :ids;59            """,60            {"package_id": main_package_id, "ids": tuple(fix_ids)},61        )62        session.execute(63            "delete from package where id in :ids",64            {"ids": tuple(fix_ids)},65        )66    log.debug("Removing indexes and constraints")67    op.drop_index('ix_package_copr_dir_id', table_name='package')68    op.drop_constraint('packages_copr_dir_pkgname', 'package', type_='unique')69    op.drop_constraint('package_copr_dir_foreign_key', 'package', type_='foreignkey')70    op.drop_column('package', 'copr_dir_id')71    # make this unique72    op.drop_index('package_copr_id_name', table_name='package')73    op.create_index('package_copr_id_name', 'package', ['copr_id', 'name'], unique=True)74def downgrade():75    op.add_column('package', sa.Column('copr_dir_id', sa.INTEGER(), autoincrement=False, nullable=True))76    op.create_foreign_key('package_copr_dir_foreign_key', 'package', 'copr_dir', ['copr_dir_id'], ['id'])77    op.create_unique_constraint('packages_copr_dir_pkgname', 'package', ['copr_dir_id', 'name'])78    op.create_index('ix_package_copr_dir_id', 'package', ['copr_dir_id'], unique=False)79    op.drop_index('package_copr_id_name', table_name='package')...

xmlexport.py

Source:xmlexport.py

1"""2Export Timetable XML from a Timetable site.3usage: ttapiutils xmlexport [options] <domain> <path>...4If more than one <path> is specified, the output will be run through5ttapiutils merge to produce a single XML file.6options:7    --user=<user>8        The username to authenticate with.9    --pass-envar=<envar>10        The name of the environment variable which contains the user's11        password. If none is provided the password will be prompted12        for on stdin.13    --https14    --no-https15        Use or don't use https [default: https].16    --fix-ids17        Fix exorted event IDs with ttapiutils fixexport (default)18    --no-fix-ids19        Don't fix event IDs20    -h, --help21        Show this help message22"""23from cStringIO import StringIO24import sys25import urlparse26from lxml import etree27from requests.exceptions import RequestException28import docopt29import requests30from ttapiutils.merge import merge31from ttapiutils.fixexport import fix_export_ids32from ttapiutils.utils import (33    write_c14n_pretty, parse_xml, assert_valid, read_password,34    get_credentials, get_proto)35class ExportException(Exception):36    pass37class HttpRequestExportException(ExportException):38    pass39class XMLParseExportException(ExportException):40    pass41def build_api_export_url(domain, path, proto="https"):42    full_path = "/api/v0/xmlexport{}".format(path)43    return urlparse.urlunparse((proto, domain, full_path, None, None, None))44def xmlexport(domain, path, auth=None, proto="https", fix_ids=True):45    url = build_api_export_url(domain, path, proto=proto)46    try:47        response = requests.get(url, auth=auth, allow_redirects=False)48        if response.status_code != requests.codes.ok:49            response.raise_for_status()50            raise HttpRequestExportException(51                "Non-200 response received to request for: {}. {}".format(52                    url, response.status_code))53    except RequestException as e:54        raise HttpRequestExportException("Error requesting timetable: {}. {}"55                                         .format(url, e))56    try:57        xml = parse_xml(StringIO(response.content))58    except etree.Error as e:59        raise XMLParseExportException(60            "Unable to parse response as XML: {}".format(e), e, response)61    assert_valid(xml)62    if fix_ids:63        xml = fix_export_ids(xml)64    return xml65def main(argv):66    args = docopt.docopt(__doc__, argv=argv)67    credentials = get_credentials(args)68    proto = get_proto(args)69    domain = args["<domain>"]70    paths = args["<path>"]71    fix_ids = not args.get("--no-fix-ids")72    exports = [xmlexport(domain, path, auth=credentials, proto=proto,73                         fix_ids=fix_ids)74               for path in paths]...

test_markdown_refs.py

Source:test_markdown_refs.py

1import os2from tests.vulndb_test import VulnDBTest3class TestReferences(VulnDBTest):4    """5    Ensure that every fix and description field has a corresponding6    markdown file, and that every markdown file is referenced by at7    least one vulnerability.8    """9    def get_references_for_language(self, language):10        desc_ids = set()11        referenced_desc_ids = set()12        fix_ids = set()13        referenced_fix_ids = set()14        for language_iter, db_path_file, db_data in self.get_all_json():15            if language_iter != language:16                continue17            desc_id = db_data['description']['$ref'].split('/')[-1]18            fix_id = db_data['fix']['guidance']['$ref'].split('/')[-1]19            referenced_desc_ids.add(desc_id)20            referenced_fix_ids.add(fix_id)21        description_path = os.path.join('db', language, 'description')22        fix_path = os.path.join('db', language, 'fix')23        for f in os.listdir(description_path):24            fpath = os.path.join(description_path, f)25            if not os.path.isfile(fpath):26                continue27            d_id = f.replace('.md', '').split('-')[0]28            desc_ids.add(d_id)29        for f in os.listdir(fix_path):30            fpath = os.path.join(fix_path, f)31            if not os.path.isfile(fpath):32                continue33            f_id = f.replace('.md', '').split('-')[0]34            fix_ids.add(f_id)35        return desc_ids, fix_ids, referenced_desc_ids, referenced_fix_ids36    def test_description_refs(self):37        for language in self.get_all_languages():38            desc_ids, _, referenced_desc_ids, _ = self.get_references_for_language(language)39            for desc_id in referenced_desc_ids:40                self.assertIn(41                    desc_id, desc_ids,42                    'description is missing: {}'.format(desc_id)43                )44            for desc_id in desc_ids:45                self.assertIn(46                    desc_id, referenced_desc_ids,47                    'description is not referenced: {}'.format(desc_id)48                )49    def test_fix_refs(self):50        for language in self.get_all_languages():51            _, fix_ids, _, referenced_fix_ids = self.get_references_for_language(language)52            for fix_id in referenced_fix_ids:53                self.assertIn(54                    fix_id, fix_ids,55                    'fix is missing: {}'.format(fix_id)56                )57            for fix_id in fix_ids:58                self.assertIn(59                    fix_id, referenced_fix_ids,60                    'fix is not referenced: {}'.format(fix_id)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.