1564245748

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/release-engineering/exodus-gw/commit/<a class=hub.com/release-engineering/exodus-gw/commit/<a class="double-link" href="https://git"><a class=hub.com/release-engineering/exodus-gw/commit/<a class="double-link" href="https://git"><a class=hub.com/release-engineering/exodus-gw/commit/adc466da6e9c2f66d09b816f3d3b2e2e41e9f0fd">adc466da6<a href="https://github.com/release-engineering/exodus-gw/commit/adc466da6e9c2f66d09b816f3d3b2e2e41e9f0fd">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/release-engineering/exodus-gw/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/release-engineering/exodus-gw/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/release-engineering/exodus-gw/commit/adc466da6e9c2f66d09b816f3d3b2e2e41e9f0fd&quot;&gt;adc466da6&lt;/a&gt;&lt;a href=&quot;https://github.com/release-engineering/exodus-gw/commit/adc466da6e9c2f66d09b816f3d3b2e2e41e9f0fd&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/release-engineering/exodus-gw/commit/adc466da6e9c2f66d09b816f3d3b2e2e&lt;/a&gt;41e9f0fd&quot;&gt;&quot;&gt;Merge &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/release-engineering/exodus-gw/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/release-engineering/exodus-gw/commit/<a class="double-link" href="https://github.com/release-engineering/exodus-gw/commit/fc68ceb6a7ecdfbddeec563f48747cf0b144eb2b">fc68ceb6a</a><a href="https://github.com/release-engineering/exodus-gw/commit/adc466da6e9c2f66d09b816f3d3b2e2e41e9f0fd">&amp;quot;&amp;gt;fc68ceb6a&amp;lt;/a&amp;gt;&amp;quot;&amp;gt;fc68ceb6a&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/release-engineering/exodus-gw/commit/adc466da6e9c2f66d09b816f3d3b2e2e41e9f&lt;/a&gt;0fd&quot;&gt; into &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/release-engineering/exodus-gw/commit/<a class="double-link" href="https://github.com/release-engineering/exodus-gw/commit/376018494">376018494">376018494</a>

Pull Request Pull Request #391: Fix: abspath check on empty/None link_to values [RHELDST-9189]

Run Details

19 of 19 new or added lines in 1 file covered. (100.0%)

1311 of 1312 relevant lines covered (99.92%)

1.0 hits per line

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

98.89

/exodus_gw/schemas.py

import re
from datetime import datetime
from enum import Enum
from os.path import abspath, join, normpath
from typing import Dict, List, Optional
from uuid import UUID

from fastapi import Path
from pydantic import BaseModel, Field, root_validator

PathPublishId = Path(
    ...,
    title="publish ID",
    description="UUID of an existing publish object.",
)

PathTaskId = Path(
    ..., title="task ID", description="UUID of an existing task object."
)


def normalize_path(path: str):
    if path:
        path = normpath(path)
        path = "/" + path if not path.startswith("/") else path
    return path


class ItemBase(BaseModel):
    web_uri: str = Field(
        ...,
        description="URI, relative to CDN root, which shall be used to expose this object.",
    )
    object_key: str = Field(
        "",
        description=(
            "Key of blob to be exposed; should be the SHA256 checksum of a previously uploaded "
            "piece of content, in lowercase hex-digest form."
        ),
    )
    link_to: str = Field("", description="Path of file targeted by symlink.")

    @root_validator()
    @classmethod
    def validate_item(cls, values):
        web_uri = values.get("web_uri")
        object_key = values.get("object_key")
        link_to = values.get("link_to")

        if not web_uri:
            raise ValueError("No URI: %s" % values)
        values["web_uri"] = normalize_path(web_uri)

        if link_to:
            if link_to != abspath(link_to):
                raise ValueError("Link target is not absolute: %s" % values)
            # Normalize last, as up-level references may be erased.
            values["link_to"] = normalize_path(link_to)
        elif object_key:
            pattern = re.compile(r"[0-9a-f]{64}")
            if not re.match(pattern, object_key):
                raise ValueError("Invalid object key: %s" % values)
        else:
            raise ValueError("No object key or link target: %s" % values)

        return values


class Item(ItemBase):
    publish_id: UUID = Field(
        ..., description="Unique ID of publish object containing this item."
    )

    class Config:
        orm_mode = True


class PublishStates(str, Enum):
    pending = "PENDING"
    committing = "COMMITTING"
    committed = "COMMITTED"
    failed = "FAILED"

    @classmethod
    def terminal(cls) -> List["PublishStates"]:
        return [cls.committed, cls.failed]


class PublishBase(BaseModel):
    id: UUID = Field(..., description="Unique ID of publish object.")


class Publish(PublishBase):
    env: str = Field(
        ..., description="""Environment to which this publish belongs."""
    )
    state: PublishStates = Field(
        ..., description="Current state of this publish."
    )
    updated: datetime = Field(
        None,
        description="DateTime of last update to this publish. None if never updated.",
    )
    links: Dict[str, str] = Field(
        {}, description="""URL links related to this publish."""
    )
    items: List[Item] = Field(
        [],
        description="""All items (pieces of content) included in this publish.""",
    )

    @root_validator
    @classmethod
    def make_links(cls, values):
        _self = join("/", values["env"], "publish", str(values["id"]))
        values["links"] = {"self": _self, "commit": join(_self, "commit")}
        return values

    class Config:
        orm_mode = True


class TaskStates(str, Enum):
    not_started = "NOT_STARTED"
    in_progress = "IN_PROGRESS"
    complete = "COMPLETE"
    failed = "FAILED"

    @classmethod
    def terminal(cls) -> List["TaskStates"]:
        return [cls.failed, cls.complete]


class Task(BaseModel):
    id: UUID = Field(..., description="Unique ID of task object.")
    publish_id: Optional[UUID] = Field(
        ..., description="Unique ID of publish object handled by this task."
    )
    state: TaskStates = Field(..., description="Current state of this task.")
    updated: datetime = Field(
        None,
        description="DateTime of last update to this task. None if never updated.",
    )
    links: Dict[str, str] = Field(
        {}, description="""URL links related to this task."""
    )

    @root_validator
    @classmethod
    def make_links(cls, values):
        values["links"] = {"self": join("/task", str(values["id"]))}
        return values

    class Config:
        orm_mode = True


class MessageResponse(BaseModel):
    detail: str = Field(
        ..., description="A human-readable message with additional info."
    )


class EmptyResponse(BaseModel):
    """An empty object."""

1	import re	1✔
2	from datetime import datetime	1✔
3	from enum import Enum	1✔
4	from os.path import abspath, join, normpath	1✔
5	from typing import Dict, List, Optional	1✔
6	from uuid import UUID	1✔
7
8	from fastapi import Path	1✔
9	from pydantic import BaseModel, Field, root_validator	1✔
10
11	PathPublishId = Path(	1✔
12	...,
13	title="publish ID",
14	description="UUID of an existing publish object.",
15	)
16
17	PathTaskId = Path(	1✔
18	..., title="task ID", description="UUID of an existing task object."
19	)
20
21
22	def normalize_path(path: str):	1✔
23	if path:	1✔
24	path = normpath(path)	1✔
25	path = "/" + path if not path.startswith("/") else path	1✔
26	return path	1✔
27
28
29	class ItemBase(BaseModel):	1✔
30	web_uri: str = Field(	1✔
31	...,
32	description="URI, relative to CDN root, which shall be used to expose this object.",
33	)
34	object_key: str = Field(	1✔
35	"",
36	description=(
37	"Key of blob to be exposed; should be the SHA256 checksum of a previously uploaded "
38	"piece of content, in lowercase hex-digest form."
39	),
40	)
41	link_to: str = Field("", description="Path of file targeted by symlink.")	1✔
42
43	@root_validator()	1✔
44	@classmethod	1✔
45	def validate_item(cls, values):	1✔
46	web_uri = values.get("web_uri")	1✔
47	object_key = values.get("object_key")	1✔
48	link_to = values.get("link_to")	1✔
49
50	if not web_uri:	1✔
51	raise ValueError("No URI: %s" % values)	1✔
52	values["web_uri"] = normalize_path(web_uri)	1✔
53
54	if link_to:	1✔
55	if link_to != abspath(link_to):	1✔
56	raise ValueError("Link target is not absolute: %s" % values)	1✔
57	# Normalize last, as up-level references may be erased.
58	values["link_to"] = normalize_path(link_to)	1✔
59	elif object_key:	1✔
60	pattern = re.compile(r"[0-9a-f]{64}")	1✔
61	if not re.match(pattern, object_key):	1✔
62	raise ValueError("Invalid object key: %s" % values)	×
63	else:
64	raise ValueError("No object key or link target: %s" % values)	1✔
65
66	return values	1✔
67
68
69	class Item(ItemBase):	1✔
70	publish_id: UUID = Field(	1✔
71	..., description="Unique ID of publish object containing this item."
72	)
73
74	class Config:	1✔
75	orm_mode = True	1✔
76
77
78	class PublishStates(str, Enum):	1✔
79	pending = "PENDING"	1✔
80	committing = "COMMITTING"	1✔
81	committed = "COMMITTED"	1✔
82	failed = "FAILED"	1✔
83
84	@classmethod	1✔
85	def terminal(cls) -> List["PublishStates"]:	1✔
86	return [cls.committed, cls.failed]	1✔
87
88
89	class PublishBase(BaseModel):	1✔
90	id: UUID = Field(..., description="Unique ID of publish object.")	1✔
91
92
93	class Publish(PublishBase):	1✔
94	env: str = Field(	1✔
95	..., description="""Environment to which this publish belongs."""
96	)
97	state: PublishStates = Field(	1✔
98	..., description="Current state of this publish."
99	)
100	updated: datetime = Field(	1✔
101	None,
102	description="DateTime of last update to this publish. None if never updated.",
103	)
104	links: Dict[str, str] = Field(	1✔
105	{}, description="""URL links related to this publish."""
106	)
107	items: List[Item] = Field(	1✔
108	[],
109	description="""All items (pieces of content) included in this publish.""",
110	)
111
112	@root_validator	1✔
113	@classmethod	1✔
114	def make_links(cls, values):	1✔
115	_self = join("/", values["env"], "publish", str(values["id"]))	1✔
116	values["links"] = {"self": _self, "commit": join(_self, "commit")}	1✔
117	return values	1✔
118
119	class Config:	1✔
120	orm_mode = True	1✔
121
122
123	class TaskStates(str, Enum):	1✔
124	not_started = "NOT_STARTED"	1✔
125	in_progress = "IN_PROGRESS"	1✔
126	complete = "COMPLETE"	1✔
127	failed = "FAILED"	1✔
128
129	@classmethod	1✔
130	def terminal(cls) -> List["TaskStates"]:	1✔
131	return [cls.failed, cls.complete]	1✔
132
133
134	class Task(BaseModel):	1✔
135	id: UUID = Field(..., description="Unique ID of task object.")	1✔
136	publish_id: Optional[UUID] = Field(	1✔
137	..., description="Unique ID of publish object handled by this task."
138	)
139	state: TaskStates = Field(..., description="Current state of this task.")	1✔
140	updated: datetime = Field(	1✔
141	None,
142	description="DateTime of last update to this task. None if never updated.",
143	)
144	links: Dict[str, str] = Field(	1✔
145	{}, description="""URL links related to this task."""
146	)
147
148	@root_validator	1✔
149	@classmethod	1✔
150	def make_links(cls, values):	1✔
151	values["links"] = {"self": join("/task", str(values["id"]))}	1✔
152	return values	1✔
153
154	class Config:	1✔
155	orm_mode = True	1✔
156
157
158	class MessageResponse(BaseModel):	1✔
159	detail: str = Field(	1✔
160	..., description="A human-readable message with additional info."
161	)
162
163
164	class EmptyResponse(BaseModel):	1✔
165	"""An empty object."""

release-engineering / exodus-gw / 1564245748

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