Completed
Push — master ( 8a9714...a444b5 )
by Jace
9s
created

TestProcesses.test_case_4()   A

Complexity

Conditions 3

Size

Total Lines 23

Duplication

Lines 0
Ratio 0 %

Importance

Changes 3
Bugs 0 Features 0
Metric Value
cc 3
dl 0
loc 23
rs 9.0856
c 3
b 0
f 0
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
    def test_data(self):
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
    def test_data_out(self):
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 self.data.status.is_running(self.application, self.computer)
188
189
        # manually mark the application as running on a remote computer
190
        computer = Computer('other', 'Other.local', 'AA:BB:CC:DD:EE:FF')
191
        status = self.data.status
192
        status.start(self.application, computer)
193
        self.data.status = status
194
        assert self.data.status.is_running(self.application, computer)
195
196
        # Act
197
198
        cli.run(self.path, cleanup=False)
199
200
        # Assert
201
202
        # verify the application is no longer running
203
        assert not self._is_application_running()
204
        # verify the application is marked as running remotely
205
        data = self._fetch_data()
206
        assert not data.status.is_running(self.application, self.computer)
207
        assert data.status.is_running(self.application, computer)
208
209
    def test_case_2(self):
210
        """Verify a newly running local application takes over."""
211
212
        # Arrange
213
214
        self._store_data()
215
216
        # manually mark the application as running on a remote computer
217
        computer = Computer('other', 'Other.local', 'AA:BB:CC:DD:EE:FF')
218
        status = self.data.status
219
        status.start(self.application, computer)
220
        self.data.status = status
221
        assert self.data.status.is_running(self.application, computer)
222
223
        # start the application
224
        self._start_application()
225
226
        # Act
227
        cli.run(self.path, cleanup=False)
228
229
        # Assert
230
231
        assert self._is_application_running()
232
233
        data = self._fetch_data()
234
        assert data.status.is_running(self.application, self.computer)
235
        assert data.status.is_running(self.application, computer)
236
237
    def test_case_3(self):
238
        """Verify an already running local application is ignored."""
239
240
        # Arrange
241
242
        self._store_data()
243
244
        status = self.data.status
245
        status.start(self.application, self.computer)
246
        self.data.status = status
247
248
        self._start_application()
249
250
        # Act
251
252
        cli.run(self.path)
253
254
        # Assert
255
256
        assert self._is_application_running()
257
258
        data = self._fetch_data()
259
        assert data.status.is_running(self.application, self.computer)
260
261
    def test_case_4(self):
262
        """Verify a newly stopped local application is noted."""
263
264
        # Arrange
265
266
        self._store_data()
267
268
        status = self.data.status
269
        status.start(self.application, self.computer)
270
        self.data.status = status
271
272
        self._stop_application()
273
274
        # Act
275
276
        cli.run(self.path)
277
278
        # Assert
279
280
        assert not self._is_application_running()
281
282
        data = self._fetch_data()
283
        assert not data.status.is_running(self.application, self.computer)
284
285
    def test_case_5(self):
286
        """Verify an already stopped local application is ignored."""
287
288
        # Arrange
289
290
        self._store_data()
291
292
        self._stop_application()
293
294
        # Act
295
296
        cli.run(self.path)
297
298
        # Assert
299
300
        assert not self._is_application_running()
301
302
        data = self._fetch_data()
303
        assert not data.status.is_running(self.application, self.computer)
304