5948296099

Committed 23 Aug 2023 07:23AM UTC coverage: 85.801% (+0.04%) from 85.766%

Build # 5948296099

Build Type

Pull #3601

github-actions

Committed by

olevski

Commit Message

chore: run poetry lock

Pull Request Pull Request #3601: hotfix: v2.6.1

Run Details

40 of 48 new or added lines in 10 files covered. (83.33%)

285 existing lines in 25 files now uncovered.

25875 of 30157 relevant lines covered (85.8%)

4.9 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.57

/renku/command/checks/workflow.py

# Copyright Swiss Data Science Center (SDSC). A partnership between
# École Polytechnique Fédérale de Lausanne (EPFL) and
# Eidgenössische Technische Hochschule Zürich (ETHZ).
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Checks needed to determine integrity of workflows."""
from datetime import timedelta
from typing import List, Optional, Tuple, cast

from renku.command.command_builder import inject
from renku.command.util import WARNING
from renku.core.interface.plan_gateway import IPlanGateway
from renku.core.util import communication
from renku.domain_model.project_context import project_context
from renku.domain_model.workflow.plan import AbstractPlan
from renku.infrastructure.gateway.activity_gateway import reindex_catalog


def check_activity_catalog(fix, force, **_) -> Tuple[bool, bool, Optional[str]]:
    """Check if the activity-catalog needs to be rebuilt.

    Args:
        fix: Whether to fix found issues.
        force: Whether to force rebuild the activity catalog.
        _: keyword arguments.

    Returns:
        Tuple of whether the activity-catalog needs to be rebuilt, if an automated fix is available and a string of
            found problems.
    """
    database = project_context.database
    activity_catalog = database["activity-catalog"]
    relations = database["_downstream_relations"]

    # NOTE: If len(activity_catalog) > 0 then either the project is fixed or it used a fixed Renku version but still has
    # broken metadata. ``force`` allows to rebuild the metadata in the latter case.
    if (len(relations) == 0 or len(activity_catalog) > 0) and not (force and fix):
        return True, False, None

    if not fix:
        problems = (
            WARNING + "The project's workflow metadata needs to be rebuilt (use 'renku doctor --fix' to rebuild it).\n"
        )

        return False, True, problems

    with communication.busy("Rebuilding workflow metadata ..."):
        reindex_catalog(database=database)

    communication.info("Workflow metadata was rebuilt")

    return True, False, None


@inject.autoparams("plan_gateway")
def check_plan_modification_date(fix, plan_gateway: IPlanGateway, **_) -> Tuple[bool, bool, Optional[str]]:
    """Check if all plans have modification date set for them.

    Args:
        fix(bool): Whether to fix found issues.
        plan_gateway(IPlanGateway): Injected PlanGateway.
        _: keyword arguments.

    Returns:
        Tuple[bool, Optional[str]]: Tuple of whether there are plans without modification date, if an automated fix is
            available and a string of their IDs
    """
    plans: List[AbstractPlan] = plan_gateway.get_all_plans()

    to_be_processed = []
    for plan in plans:
        if not hasattr(plan, "date_modified") or plan.date_modified is None:
            to_be_processed.append(plan)

    if not to_be_processed:
        return True, False, None
    if not fix:
        ids = [plan.id for plan in to_be_processed]
        message = (
            WARNING
            + "The following workflows have incorrect modification date (use 'renku doctor --fix' to fix them):\n\t"
            + "\n\t".join(ids)
        )
        return False, True, message

    fix_plan_dates(plans=to_be_processed, plan_gateway=plan_gateway)
    project_context.database.commit()
    communication.info("Workflow modification dates were fixed")

    return True, False, None


def fix_plan_dates(plans: List[AbstractPlan], plan_gateway):
    """Set modification date on a list of plans and fix their creation date."""
    processed = set()
    # NOTE: switch creation date for modification date
    for tail in plans:
        to_be_processed: List[AbstractPlan] = []
        if tail not in processed:
            processed.add(tail)
            to_be_processed.append(tail)
        creation_date = tail.date_created
        plan = tail

        while plan.is_derivation():
            plan = cast(AbstractPlan, plan_gateway.get_by_id(plan.derived_from))
            creation_date = plan.date_created
            if plan not in processed:
                processed.add(plan)
                to_be_processed.append(plan)

        while to_be_processed:
            plan = to_be_processed.pop()
            plan.unfreeze()
            plan.date_modified = plan.date_created
            plan.date_created = creation_date
            if plan.date_removed and plan.date_removed < plan.date_created:
                plan.date_removed = plan.date_created + timedelta(seconds=1)
            plan.freeze()


@inject.autoparams("plan_gateway")
def check_plan_id(fix, plan_gateway: IPlanGateway, **_) -> Tuple[bool, bool, Optional[str]]:
    """Check if all plans have correct IDs.

    Args:
        fix(bool): Whether to fix found issues.
        plan_gateway(IPlanGateway): Injected PlanGateway.
        _: keyword arguments.

    Returns:
        Tuple[bool, Optional[str]]: Tuple of whether there are plans with invalid IDs, if an automated fix is
            available and a string of their IDs
    """
    plans: List[AbstractPlan] = plan_gateway.get_all_plans()

    to_be_processed = []
    for plan in plans:
        if isinstance(plan.id, str) and plan.id.startswith("/plans//plans"):
            to_be_processed.append(plan)

    if not to_be_processed:
        return True, False, None
    if not fix:
        ids = [plan.id for plan in to_be_processed]
        message = (
            WARNING
            + "The following workflows have incorrect IDs (use 'renku doctor --fix' to fix them):\n\t"
            + "\n\t".join(ids)
        )
        return False, True, message

    for plan in to_be_processed:
        plan.unfreeze()
        plan.id = plan.id.replace("//plans/", "/")
        plan.freeze()
    project_context.database.commit()
    communication.info("Workflow IDs were fixed")

    return True, False, None

1	# Copyright Swiss Data Science Center (SDSC). A partnership between
2	# École Polytechnique Fédérale de Lausanne (EPFL) and
3	# Eidgenössische Technische Hochschule Zürich (ETHZ).
4	#
5	# Licensed under the Apache License, Version 2.0 (the "License");
6	# you may not use this file except in compliance with the License.
7	# You may obtain a copy of the License at
8	#
9	# http://www.apache.org/licenses/LICENSE-2.0
10	#
11	# Unless required by applicable law or agreed to in writing, software
12	# distributed under the License is distributed on an "AS IS" BASIS,
13	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	# See the License for the specific language governing permissions and
15	# limitations under the License.
16	"""Checks needed to determine integrity of workflows."""	5✔
17	from datetime import timedelta	5✔
18	from typing import List, Optional, Tuple, cast	5✔
19
20	from renku.command.command_builder import inject	5✔
21	from renku.command.util import WARNING	5✔
22	from renku.core.interface.plan_gateway import IPlanGateway	5✔
23	from renku.core.util import communication	5✔
24	from renku.domain_model.project_context import project_context	5✔
25	from renku.domain_model.workflow.plan import AbstractPlan	5✔
26	from renku.infrastructure.gateway.activity_gateway import reindex_catalog	5✔
27
28
29	def check_activity_catalog(fix, force, **_) -> Tuple[bool, bool, Optional[str]]:	5✔
30	"""Check if the activity-catalog needs to be rebuilt.
31
32	Args:
33	fix: Whether to fix found issues.
34	force: Whether to force rebuild the activity catalog.
35	_: keyword arguments.
36
37	Returns:
38	Tuple of whether the activity-catalog needs to be rebuilt, if an automated fix is available and a string of
39	found problems.
40	"""
41	database = project_context.database	3✔
42	activity_catalog = database["activity-catalog"]	3✔
43	relations = database["_downstream_relations"]	3✔
44
45	# NOTE: If len(activity_catalog) > 0 then either the project is fixed or it used a fixed Renku version but still has
46	# broken metadata. ``force`` allows to rebuild the metadata in the latter case.
47	if (len(relations) == 0 or len(activity_catalog) > 0) and not (force and fix):	3✔
48	return True, False, None	3✔
49
50	if not fix:	2✔
51	problems = (	1✔
52	WARNING + "The project's workflow metadata needs to be rebuilt (use 'renku doctor --fix' to rebuild it).\n"
53	)
54
55	return False, True, problems	1✔
56
57	with communication.busy("Rebuilding workflow metadata ..."):	2✔
58	reindex_catalog(database=database)	2✔
59
60	communication.info("Workflow metadata was rebuilt")	2✔
61
62	return True, False, None	2✔
63
64
65	@inject.autoparams("plan_gateway")	5✔
66	def check_plan_modification_date(fix, plan_gateway: IPlanGateway, **_) -> Tuple[bool, bool, Optional[str]]:	5✔
67	"""Check if all plans have modification date set for them.
68
69	Args:
70	fix(bool): Whether to fix found issues.
71	plan_gateway(IPlanGateway): Injected PlanGateway.
72	_: keyword arguments.
73
74	Returns:
75	Tuple[bool, Optional[str]]: Tuple of whether there are plans without modification date, if an automated fix is
76	available and a string of their IDs
77	"""
78	plans: List[AbstractPlan] = plan_gateway.get_all_plans()	3✔
79
80	to_be_processed = []	3✔
81	for plan in plans:	3✔
82	if not hasattr(plan, "date_modified") or plan.date_modified is None:	2✔
83	to_be_processed.append(plan)	1✔
84
85	if not to_be_processed:	3✔
86	return True, False, None	3✔
87	if not fix:	1✔
88	ids = [plan.id for plan in to_be_processed]	×
89	message = (	×
90	WARNING
91	+ "The following workflows have incorrect modification date (use 'renku doctor --fix' to fix them):\n\t"
92	+ "\n\t".join(ids)
93	)
94	return False, True, message	×
95
96	fix_plan_dates(plans=to_be_processed, plan_gateway=plan_gateway)	1✔
97	project_context.database.commit()	1✔
98	communication.info("Workflow modification dates were fixed")	1✔
99
100	return True, False, None	1✔
101
102
103	def fix_plan_dates(plans: List[AbstractPlan], plan_gateway):	5✔
104	"""Set modification date on a list of plans and fix their creation date."""
105	processed = set()	4✔
106	# NOTE: switch creation date for modification date
107	for tail in plans:	4✔
108	to_be_processed: List[AbstractPlan] = []	3✔
109	if tail not in processed:	3✔
110	processed.add(tail)	3✔
111	to_be_processed.append(tail)	3✔
112	creation_date = tail.date_created	3✔
113	plan = tail	3✔
114
115	while plan.is_derivation():	3✔
116	plan = cast(AbstractPlan, plan_gateway.get_by_id(plan.derived_from))	2✔
117	creation_date = plan.date_created	2✔
118	if plan not in processed:	2✔
119	processed.add(plan)	2✔
120	to_be_processed.append(plan)	2✔
121
122	while to_be_processed:	3✔
123	plan = to_be_processed.pop()	3✔
124	plan.unfreeze()	3✔
125	plan.date_modified = plan.date_created	3✔
126	plan.date_created = creation_date	3✔
127	if plan.date_removed and plan.date_removed < plan.date_created:	3✔
128	plan.date_removed = plan.date_created + timedelta(seconds=1)	×
129	plan.freeze()	3✔
130
131
132	@inject.autoparams("plan_gateway")	5✔
133	def check_plan_id(fix, plan_gateway: IPlanGateway, **_) -> Tuple[bool, bool, Optional[str]]:	5✔
134	"""Check if all plans have correct IDs.
135
136	Args:
137	fix(bool): Whether to fix found issues.
138	plan_gateway(IPlanGateway): Injected PlanGateway.
139	_: keyword arguments.
140
141	Returns:
142	Tuple[bool, Optional[str]]: Tuple of whether there are plans with invalid IDs, if an automated fix is
143	available and a string of their IDs
144	"""
145	plans: List[AbstractPlan] = plan_gateway.get_all_plans()	3✔
146
147	to_be_processed = []	3✔
148	for plan in plans:	3✔
149	if isinstance(plan.id, str) and plan.id.startswith("/plans//plans"):	2✔
150	to_be_processed.append(plan)	1✔
151
152	if not to_be_processed:	3✔
153	return True, False, None	3✔
154	if not fix:	1✔
NEW 155	ids = [plan.id for plan in to_be_processed]	×
NEW 156	message = (	×
157	WARNING
158	+ "The following workflows have incorrect IDs (use 'renku doctor --fix' to fix them):\n\t"
159	+ "\n\t".join(ids)
160	)
NEW 161	return False, True, message	×
162
163	for plan in to_be_processed:	1✔
164	plan.unfreeze()	1✔
165	plan.id = plan.id.replace("//plans/", "/")	1✔
166	plan.freeze()	1✔
167	project_context.database.commit()	1✔
168	communication.info("Workflow IDs were fixed")	1✔
169
170	return True, False, None	1✔

SwissDataScienceCenter / renku-python / 5948296099

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous