Skip to content

load_test_data#

Command#

Bases: BaseCommand

Source code in src/apps/core/management/commands/load_test_data.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
class Command(BaseCommand):
    def handle(self, *args, **options):
        if settings.ENV == "production":
            logger.warn("This command can only be used in non-production environments")
            return

        language = factories.LanguageFactory()
        data_catalog = factories.DataCatalogFactory(
            id="urn:nbn:fi:att:data-catalog-test", allowed_pid_types=["URN", "DOI"]
        )
        harvested_data_catalog = factories.DataCatalogFactory(
            id="urn:nbn:fi:att:data-catalog-harvested-test", harvested="True"
        )
        dataset = None
        files = None
        if not Dataset.objects.filter(title={"en": "Test dataset"}).exists():
            dataset = factories.DatasetFactory(
                data_catalog=data_catalog, title={"en": "Test dataset"}
            )
            logger.info(f"{dataset=}")

            files = file_factories.create_project_with_files(
                file_paths=[
                    "/dir/sub1/file1.csv",
                    "/dir/a.txt",
                    "/rootfile.txt",
                ],
                file_args={"*": {"size": 1024}},
                storage_service="test",
            )
            file_set = factories.FileSetFactory(
                dataset=dataset, storage=files["storage"], files=files["files"].values()
            )

        # Contract needed by IDA tests
        contract = Contract.objects.get_or_create(
            id="495611c1-22b2-4095-bac6-2bf676242a4f",
            defaults={
                "title": {"en": "Test contract"},
                "quota": 123456789,
                "validity_start_date": "2023-06-15",
                "validity_end_date": "2043-12-31",
                "created": "2021-12-31T12:13:14Z",
                "modified": "2021-12-31T12:13:14Z",
            },
        )

        logger.info(
            f"Created or updated test objects:\n {language=}, \n {data_catalog=}, \n {harvested_data_catalog=},"
            f"\n {dataset=},\n {files=},\n {contract=}"
        )

        self.create_data_catalogs()

        self.stdout.write("test objects created successfully")

    def create_data_catalogs(self):
        with open(
            "src/apps/core/management/initial_data/initial_data_catalogs.json", "r"
        ) as catalogs_file:
            catalogs_list = json.load(catalogs_file)
            for catalog_json in catalogs_list:  # pragma: no cover
                if data_catalog := DataCatalog.objects.filter(id=catalog_json["id"]).first():
                    serializer = DataCatalogModelSerializer(data_catalog, data=catalog_json)
                    serializer.is_valid()
                    catalog = serializer.update(
                        data_catalog, validated_data=serializer.validated_data
                    )
                    logger.info(f"Updated data catalog: {catalog}")
                    continue

                serializer = DataCatalogModelSerializer(data=catalog_json)
                valid = serializer.is_valid()
                if valid:
                    catalog = serializer.create(validated_data=serializer.validated_data)
                    logger.info(f"Created data catalog: {catalog}")
                else:
                    logger.error(f"Catalog JSON not valid: {serializer.errors}")

            logger.info(f"Created or updated test data-catalogs")

            self.stdout.write("test catalogs created successfully")