| 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 |