forked from altitudenetworks/newversion
-
Notifications
You must be signed in to change notification settings - Fork 2
/
executor.py
207 lines (168 loc) · 6.32 KB
/
executor.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
"""
CLI commands executor.
"""
import operator
from pathlib import Path
from typing import Optional
from newversion.constants import VersionParts
from newversion.exceptions import ExecutorError, PackageVersionError
from newversion.package_version import PackageVersion
from newversion.type_defs import OperatorTypeDef, ReleaseNonLocalTypeDef, ReleaseTypeDef
from newversion.version import Version
class Executor:
"""
CLI commands executor.
"""
def __init__(
self,
input: Optional[Version] = None,
) -> None:
self._input = input if input is not None else Version.zero()
def command_get(
self,
release: ReleaseTypeDef,
) -> str:
"""
Get version part.
Arguments:
release -- Release part name.
Returns:
Part as a string.
"""
if release == VersionParts.LOCAL:
return self._input.local if self._input.local else ""
if release == VersionParts.PRE:
return f"{self._input.pre[0]}{self._input.pre[1]}" if self._input.pre else ""
if release == VersionParts.POST:
return str(self._input.post) if self._input.post else "0"
if release == VersionParts.DEV:
return str(self._input.dev) if self._input.dev else "0"
if release == VersionParts.ALPHA:
return (
str(self._input.pre[-1]) if self._input.pre and self._input.pre[0] == "a" else "0"
)
if release == VersionParts.BETA:
return (
str(self._input.pre[-1]) if self._input.pre and self._input.pre[0] == "b" else "0"
)
if release == VersionParts.RC:
return (
str(self._input.pre[-1]) if self._input.pre and self._input.pre[0] == "rc" else "0"
)
if release == VersionParts.EPOCH:
return str(self._input.epoch) if self._input.epoch else "0"
result = dict(
major=self._input.major,
minor=self._input.minor,
micro=self._input.micro,
)[release]
return str(result)
def command_bump(self, release: ReleaseNonLocalTypeDef, increment: int) -> Version:
"""
Bump release.
Arguments:
release -- Release name
increment -- Number to increase by
Returns:
A new Version.
"""
if (
release == VersionParts.MAJOR
or release == VersionParts.MINOR
or release == VersionParts.MICRO
):
return self._input.bump_release(release, increment)
if release == VersionParts.PRE:
return self._input.bump_prerelease(increment)
if release == VersionParts.POST:
return self._input.bump_postrelease(increment)
if (
release == VersionParts.RC
or release == VersionParts.ALPHA
or release == VersionParts.BETA
):
return self._input.bump_prerelease(increment, release)
if release == VersionParts.DEV:
return self._input.bump_dev(increment)
raise ExecutorError(f"Unknown release name: {release}")
def command_set(self, release: ReleaseNonLocalTypeDef, value: int) -> Version:
"""
Set version part.
Arguments:
release -- Release name
value -- Value to set
Returns:
A new Version.
"""
if release == VersionParts.PRE:
if self._input.prerelease_type == VersionParts.ALPHA:
return self._input.replace(alpha=value)
if self._input.prerelease_type == VersionParts.BETA:
return self._input.replace(beta=value)
if self._input.prerelease_type == VersionParts.RC:
return self._input.replace(rc=value)
return self._input.replace(rc=value)
if release == VersionParts.POST:
return self._input.replace(post=value)
if release == VersionParts.EPOCH:
return self._input.replace(epoch=value)
if release == VersionParts.MAJOR:
return self._input.replace(major=value)
if release == VersionParts.MINOR:
return self._input.replace(minor=value)
if release == VersionParts.MICRO:
return self._input.replace(micro=value)
if release == VersionParts.ALPHA:
return self._input.replace(alpha=value)
if release == VersionParts.BETA:
return self._input.replace(beta=value)
if release == VersionParts.RC:
return self._input.replace(rc=value)
if release == VersionParts.DEV:
return self._input.replace(dev=value)
return self._input
def command_stable(self) -> Version:
"""
Get stable non-post, non-local version from current.
Returns:
A new Version.
"""
return self._input.get_stable()
def command_is_stable(self) -> None:
"""
Check whether version is stable.
Raises:
ExecutorError -- If it is not.
"""
if not self._input.is_stable:
raise ExecutorError(f"Version {self._input} is not stable")
def command_compare(self, command: OperatorTypeDef, other: Version) -> None:
"""
Execute compare command.
Arguments:
command -- Compare operator.
other -- Version to compare to.
Returns:
Processed `Version`.
"""
commands = dict(
lt=(operator.lt, "not lesser than"),
lte=(operator.le, "not lesser than or equal to"),
gt=(operator.gt, "not greater than"),
gte=(operator.ge, "not greater than or equal to"),
eq=(operator.eq, "not equal to"),
ne=(operator.ne, "equal to"),
)
op, message = commands[command]
if not (op(self._input, other)):
raise ExecutorError(f"Version {self._input} is {message} {other}")
def command_get_version(self) -> Version:
try:
return PackageVersion(Path.cwd()).get()
except PackageVersionError as e:
raise ExecutorError(e)
def command_set_version(self) -> None:
try:
PackageVersion(Path.cwd()).set(self._input)
except PackageVersionError as e:
raise ExecutorError(e)