Total Complexity | 40 |
Total Lines | 194 |
Duplicated Lines | 25.77 % |
Changes | 3 | ||
Bugs | 0 | Features | 0 |
Duplicate code is one of the most pungent code smells. A rule that is often used is to re-structure code once it is duplicated in three or more places.
Common duplication problems, and corresponding solutions are:
Complex classes like TestProcesses often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.
Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.
1 | # pylint: disable=misplaced-comparison-constant,no-self-use |
||
120 | class TestProcesses: |
||
121 | """Integration tests for tracking and stopping processes.""" |
||
122 | |||
123 | NAME = "example application" |
||
124 | |||
125 | application = Application(NAME) |
||
126 | application.versions.linux = 'yes' |
||
127 | application.versions.mac = 'yes' |
||
128 | application.versions.windows = 'yes.exe' |
||
129 | |||
130 | computer = None |
||
131 | data = None |
||
132 | |||
133 | path = os.path.join(FILES, 'mine.tmp.yml') |
||
134 | |||
135 | _process = None |
||
136 | |||
137 | def _store_data(self): |
||
138 | """Set up initial data file for tests.""" |
||
139 | self.data = Data() |
||
140 | self.data.config.applications.append(self.application) |
||
141 | self.computer = self.data.config.computers.get_current() # pylint: disable=no-member |
||
142 | yorm.sync(self.data, self.path) |
||
143 | |||
144 | def _fetch_data(self): |
||
145 | """Read the final data file back for verification.""" |
||
146 | data = Data() |
||
147 | yorm.sync(data, self.path) |
||
148 | return data |
||
149 | |||
150 | def _start_application(self): |
||
151 | """Start the example application.""" |
||
152 | if not self._process: |
||
153 | # TODO: get filename from the application object |
||
154 | self._process = subprocess.Popen(['yes'], stdout=subprocess.PIPE) |
||
155 | log.info("%s is started", self.application) |
||
156 | |||
157 | def _stop_application(self): |
||
158 | """Stop the example application.""" |
||
159 | if self._process: |
||
160 | if self._process.poll() is None: |
||
161 | self._process.kill() |
||
162 | self._process = None |
||
163 | log.info("%s is stopped", self.application) |
||
164 | |||
165 | def _is_application_running(self): |
||
166 | """Determine if the sample application is running.""" |
||
167 | return self._process and self._process.poll() is None |
||
168 | |||
169 | def teardown_method(self, _): |
||
170 | """Stop the sample application and clean up the file.""" |
||
171 | self._stop_application() |
||
172 | if os.path.exists(self.path): |
||
173 | os.remove(self.path) |
||
174 | |||
175 | def test_case_1(self): |
||
176 | """Verify a newly running remote application takes over.""" |
||
177 | |||
178 | # Arrange |
||
179 | |||
180 | self._store_data() |
||
181 | |||
182 | # start the application and manually mark it as running |
||
183 | self._start_application() |
||
184 | status = self.data.status |
||
185 | status.start(self.application, self.computer) |
||
186 | self.data.status = status |
||
187 | assert 1 == self.data.status.counter |
||
188 | assert self.data.status.is_running(self.application, self.computer) |
||
189 | |||
190 | # manually mark the application as running on a remote computer |
||
191 | computer = Computer('other', 'Other.local', 'AA:BB:CC:DD:EE:FF') |
||
192 | status = self.data.status |
||
193 | status.start(self.application, computer) |
||
194 | self.data.status = status |
||
195 | assert 2 == self.data.status.counter |
||
196 | assert self.data.status.is_running(self.application, computer) |
||
197 | |||
198 | # Act |
||
199 | |||
200 | cli.run(self.path, cleanup=False) |
||
201 | |||
202 | # Assert |
||
203 | |||
204 | # verify the application is no longer running |
||
205 | assert not self._is_application_running() |
||
206 | # verify the application is marked as running remotely |
||
207 | data = self._fetch_data() |
||
208 | assert 3 == data.status.counter |
||
209 | assert not data.status.is_running(self.application, self.computer) |
||
210 | assert data.status.is_running(self.application, computer) |
||
211 | |||
212 | def test_case_2(self): |
||
213 | """Verify a newly running local application takes over.""" |
||
214 | |||
215 | # Arrange |
||
216 | |||
217 | self._store_data() |
||
218 | |||
219 | # manually mark the application as running on a remote computer |
||
220 | computer = Computer('other', 'Other.local', 'AA:BB:CC:DD:EE:FF') |
||
221 | status = self.data.status |
||
222 | status.start(self.application, computer) |
||
223 | self.data.status = status |
||
224 | assert 1 == self.data.status.counter |
||
225 | assert self.data.status.is_running(self.application, computer) |
||
226 | |||
227 | # start the application |
||
228 | self._start_application() |
||
229 | |||
230 | # Act |
||
231 | cli.run(self.path, cleanup=False) |
||
232 | |||
233 | # Assert |
||
234 | |||
235 | assert self._is_application_running() |
||
236 | |||
237 | data = self._fetch_data() |
||
238 | assert 2 == data.status.counter |
||
239 | assert data.status.is_running(self.application, self.computer) |
||
240 | assert data.status.is_running(self.application, computer) |
||
241 | |||
242 | View Code Duplication | def test_case_3(self): |
|
243 | """Verify an already running local application is ignored.""" |
||
244 | |||
245 | # Arrange |
||
246 | |||
247 | self._store_data() |
||
248 | |||
249 | status = self.data.status |
||
250 | status.start(self.application, self.computer) |
||
251 | self.data.status = status |
||
252 | assert 1 == self.data.status.counter |
||
253 | |||
254 | self._start_application() |
||
255 | |||
256 | # Act |
||
257 | |||
258 | cli.run(self.path) |
||
259 | |||
260 | # Assert |
||
261 | |||
262 | assert self._is_application_running() |
||
263 | |||
264 | data = self._fetch_data() |
||
265 | assert 1 == data.status.counter |
||
266 | assert data.status.is_running(self.application, self.computer) |
||
267 | |||
268 | View Code Duplication | def test_case_4(self): |
|
269 | """Verify a newly stopped local application is noted.""" |
||
270 | |||
271 | # Arrange |
||
272 | |||
273 | self._store_data() |
||
274 | |||
275 | status = self.data.status |
||
276 | status.start(self.application, self.computer) |
||
277 | self.data.status = status |
||
278 | assert 1 == self.data.status.counter |
||
279 | |||
280 | self._stop_application() |
||
281 | |||
282 | # Act |
||
283 | |||
284 | cli.run(self.path) |
||
285 | |||
286 | # Assert |
||
287 | |||
288 | assert not self._is_application_running() |
||
289 | |||
290 | data = self._fetch_data() |
||
291 | assert 2 == data.status.counter |
||
292 | assert not data.status.is_running(self.application, self.computer) |
||
293 | |||
294 | def test_case_5(self): |
||
295 | """Verify an already stopped local application is ignored.""" |
||
296 | |||
297 | # Arrange |
||
298 | |||
299 | self._store_data() |
||
300 | |||
301 | self._stop_application() |
||
302 | |||
303 | # Act |
||
304 | |||
305 | cli.run(self.path) |
||
306 | |||
307 | # Assert |
||
308 | |||
309 | assert not self._is_application_running() |
||
310 | |||
311 | data = self._fetch_data() |
||
312 | assert 0 == data.status.counter |
||
313 | assert not data.status.is_running(self.application, self.computer) |
||
314 |