Completed
Pull Request — master (#58)
by Jace
02:38
created

TestProcesses.test_case_1()   F

Complexity

Conditions 9

Size

Total Lines 36

Duplication

Lines 0
Ratio 0 %

Importance

Changes 3
Bugs 0 Features 0
Metric Value
cc 9
c 3
b 0
f 0
dl 0
loc 36
rs 3
1
# pylint: disable=misplaced-comparison-constant,no-self-use
0 ignored issues
show
introduced by
Bad option value 'misplaced-comparison-constant'
Loading history...
2
3
import os
4
import subprocess
5
import logging
6
7
import yorm
0 ignored issues
show
Configuration introduced by
The import yorm could not be resolved.

This can be caused by one of the following:

1. Missing Dependencies

This error could indicate a configuration issue of Pylint. Make sure that your libraries are available by adding the necessary commands.

# .scrutinizer.yml
before_commands:
    - sudo pip install abc # Python2
    - sudo pip3 install abc # Python3
Tip: We are currently not using virtualenv to run pylint, when installing your modules make sure to use the command for the correct version.

2. Missing __init__.py files

This error could also result from missing __init__.py files in your module folders. Make sure that you place one file in each sub-folder.

Loading history...
8
9
from mine import cli
10
from mine.application import Application
11
from mine.computer import Computer
12
from mine.status import State, Status, ProgramStatus
13
from mine.config import ProgramConfig
14
from mine.data import Data
15
16
from .conftest import FILES
17
18
19
log = logging.getLogger(__name__)
20
21
22
class TestFiles:
23
    """Integration tests for creating files."""
24
25 View Code Duplication
    def test_data(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
26
        """Verify a sample file is created."""
27
        path = os.path.join(FILES, 'mine.yml')
28
29
        if os.path.exists(path):
30
            os.remove(path)
31
32
        data = Data()
33
        yorm.sync(data, path)
34
35
        itunes = Application('itunes')
36
        itunes.versions.mac = ''
37
        itunes.versions.windows = 'iTunes.exe'
38
39
        iphoto = Application('iphoto')
40
        iphoto.versions.mac = 'iPhoto'
41
42
        mac = Computer('macbook', 'Other.local')
43
        mac2 = Computer('macbook-pro')
44
45
        configuration = ProgramConfig()
46
        configuration.applications = [itunes, iphoto]
47
        configuration.computers = [mac, mac2]
48
49
        data.config = configuration
50
51
        mac_state = State('macbook-pro')
52
        mac_state.timestamp.started = 444
53
54
        itunes_status = Status('itunes')
55
        itunes_status.computers = [mac_state]
56
57
        status = ProgramStatus()
58
        status.applications = [itunes_status]
59
        status.counter = 499
60
61
        data.status = status
62
63
        assert os.path.exists(path)
64
65 View Code Duplication
    def test_data_out(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
66
        """Verify a sample file is created."""
67
        path = os.path.join(FILES, 'mine-out.yml')
68
69
        if os.path.exists(path):
70
            os.remove(path)
71
72
        data = Data()
73
        yorm.sync(data, path)
74
75
        itunes = Application('itunes')
76
        itunes.versions.mac = ''
77
        itunes.versions.windows = 'iTunes.exe'
78
79
        iphoto = Application('iphoto')
80
        iphoto.versions.mac = 'iPhoto'
81
82
        mac = Computer('macbook', 'Jaces-MacBook', 'AA:BB:CC:DD:EE:FF')
83
        mac2 = Computer('macbook-pro', 'Jaces-MacBook-2', '11:22:33:44:55:66')
84
85
        configuration = ProgramConfig()
86
        configuration.applications = [itunes, iphoto]
87
        configuration.computers = [mac, mac2]
88
89
        data.config = configuration
90
91
        mac_state = State('macbook-pro')
92
        mac_state.timestamp.started = 444
93
94
        itunes_status = Status('itunes')
95
        itunes_status.computers = [mac_state]
96
97
        status = ProgramStatus()
98
        status.applications = [itunes_status]
99
        status.counter = 499
100
101
        data.status = status
102
103
        assert os.path.exists(path)
104
105
    def test_data_in(self):
106
        """Verify a sample file is loaded."""
107
        path = os.path.join(FILES, 'mine-in.yml')
108
109
        data = Data()
110
        yorm.sync(data, path)
111
112
        assert data.config.applications
113
        for application in data.config.applications:
114
            if application.name == 'slack':
115
                break
116
        else:
117
            assert False
118
119
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):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
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):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
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