Completed
Pull Request — master (#68)
by Jace
03:27
created

tests/test_all.py (2 issues)

1
# pylint: disable=misplaced-comparison-constant,no-self-use
2
3
import os
4
import subprocess
5
import logging
6
7
import yorm
8
9
from mine import cli
10
from mine.models import (Application, Computer, Data, ProgramConfig,
11
                         State, Status, ProgramStatus)
12
13
from .conftest import FILES
14
15
16
log = logging.getLogger(__name__)
17
18
19
class TestFiles:
20
    """Integration tests for creating files."""
21
22 View Code Duplication
    def test_data(self):
0 ignored issues
show
This code seems to be duplicated in your project.
Loading history...
23
        """Verify a sample file is created."""
24
        path = os.path.join(FILES, 'mine.yml')
25
26
        if os.path.exists(path):
27
            os.remove(path)
28
29
        data = Data()
30
        yorm.sync(data, path)
31
32
        itunes = Application('itunes')
33
        itunes.versions.mac = ''
34
        itunes.versions.windows = 'iTunes.exe'
35
36
        iphoto = Application('iphoto')
37
        iphoto.versions.mac = 'iPhoto'
38
39
        mac = Computer('macbook', 'Other.local')
40
        mac2 = Computer('macbook-pro')
41
42
        configuration = ProgramConfig()
43
        configuration.applications = [itunes, iphoto]
44
        configuration.computers = [mac, mac2]
45
46
        data.config = configuration
47
48
        mac_state = State('macbook-pro')
49
        mac_state.timestamp.started = 444
50
51
        itunes_status = Status('itunes')
52
        itunes_status.computers = [mac_state]
53
54
        status = ProgramStatus()
55
        status.applications = [itunes_status]
56
        status.counter = 499
57
58
        data.status = status
59
60
        assert os.path.exists(path)
61
62 View Code Duplication
    def test_data_out(self):
0 ignored issues
show
This code seems to be duplicated in your project.
Loading history...
63
        """Verify a sample file is created."""
64
        path = os.path.join(FILES, 'mine-out.yml')
65
66
        if os.path.exists(path):
67
            os.remove(path)
68
69
        data = Data()
70
        yorm.sync(data, path)
71
72
        itunes = Application('itunes')
73
        itunes.versions.mac = ''
74
        itunes.versions.windows = 'iTunes.exe'
75
76
        iphoto = Application('iphoto')
77
        iphoto.versions.mac = 'iPhoto'
78
79
        mac = Computer('macbook', 'Jaces-MacBook', 'AA:BB:CC:DD:EE:FF')
80
        mac2 = Computer('macbook-pro', 'Jaces-MacBook-2', '11:22:33:44:55:66')
81
82
        configuration = ProgramConfig()
83
        configuration.applications = [itunes, iphoto]
84
        configuration.computers = [mac, mac2]
85
86
        data.config = configuration
87
88
        mac_state = State('macbook-pro')
89
        mac_state.timestamp.started = 444
90
91
        itunes_status = Status('itunes')
92
        itunes_status.computers = [mac_state]
93
94
        status = ProgramStatus()
95
        status.applications = [itunes_status]
96
        status.counter = 499
97
98
        data.status = status
99
100
        assert os.path.exists(path)
101
102
    def test_data_in(self):
103
        """Verify a sample file is loaded."""
104
        path = os.path.join(FILES, 'mine-in.yml')
105
106
        data = Data()
107
        yorm.sync(data, path)
108
109
        assert data.config.applications
110
        for application in data.config.applications:
111
            if application.name == 'slack':
112
                break
113
        else:
114
            assert False
115
116
117
class TestProcesses:
118
    """Integration tests for tracking and stopping processes."""
119
120
    NAME = "example application"
121
122
    application = Application(NAME)
123
    application.versions.linux = 'yes'
124
    application.versions.mac = 'yes'
125
    application.versions.windows = 'yes.exe'
126
127
    computer = None
128
    data = None
129
130
    path = os.path.join(FILES, 'mine.tmp.yml')
131
132
    _process = None
133
134
    def _store_data(self):
135
        """Set up initial data file for tests."""
136
        self.data = Data()
137
        self.data.config.applications.append(self.application)
138
        self.computer = self.data.config.computers.get_current()  # pylint: disable=no-member
139
        yorm.sync(self.data, self.path)
140
141
    def _fetch_data(self):
142
        """Read the final data file back for verification."""
143
        data = Data()
144
        yorm.sync(data, self.path)
145
        return data
146
147
    def _start_application(self):
148
        """Start the example application."""
149
        if not self._process:
150
            # TODO: get filename from the application object
151
            self._process = subprocess.Popen(['yes'], stdout=subprocess.PIPE)
152
        log.info("%s is started", self.application)
153
154
    def _stop_application(self):
155
        """Stop the example application."""
156
        if self._process:
157
            if self._process.poll() is None:
158
                self._process.kill()
159
            self._process = None
160
        log.info("%s is stopped", self.application)
161
162
    def _is_application_running(self):
163
        """Determine if the sample application is running."""
164
        return self._process and self._process.poll() is None
165
166
    def teardown_method(self, _):
167
        """Stop the sample application and clean up the file."""
168
        self._stop_application()
169
        if os.path.exists(self.path):
170
            os.remove(self.path)
171
172
    def test_case_1(self):
173
        """Verify a newly running remote application takes over."""
174
175
        # Arrange
176
177
        self._store_data()
178
179
        # start the application and manually mark it as running
180
        self._start_application()
181
        status = self.data.status
182
        status.start(self.application, self.computer)
183
        self.data.status = status
184
        assert self.data.status.is_running(self.application, self.computer)
185
186
        # manually mark the application as running on a remote computer
187
        computer = Computer('other', 'Other.local', 'AA:BB:CC:DD:EE:FF')
188
        status = self.data.status
189
        status.start(self.application, computer)
190
        self.data.status = status
191
        assert self.data.status.is_running(self.application, computer)
192
193
        # Act
194
195
        cli.run(self.path, cleanup=False)
196
197
        # Assert
198
199
        # verify the application is no longer running
200
        assert not self._is_application_running()
201
        # verify the application is marked as running remotely
202
        data = self._fetch_data()
203
        assert not data.status.is_running(self.application, self.computer)
204
        assert data.status.is_running(self.application, computer)
205
206
    def test_case_2(self):
207
        """Verify a newly running local application takes over."""
208
209
        # Arrange
210
211
        self._store_data()
212
213
        # manually mark the application as running on a remote computer
214
        computer = Computer('other', 'Other.local', 'AA:BB:CC:DD:EE:FF')
215
        status = self.data.status
216
        status.start(self.application, computer)
217
        self.data.status = status
218
        assert self.data.status.is_running(self.application, computer)
219
220
        # start the application
221
        self._start_application()
222
223
        # Act
224
        cli.run(self.path, cleanup=False)
225
226
        # Assert
227
228
        assert self._is_application_running()
229
230
        data = self._fetch_data()
231
        assert data.status.is_running(self.application, self.computer)
232
        assert data.status.is_running(self.application, computer)
233
234
    def test_case_3(self):
235
        """Verify an already running local application is ignored."""
236
237
        # Arrange
238
239
        self._store_data()
240
241
        status = self.data.status
242
        status.start(self.application, self.computer)
243
        self.data.status = status
244
245
        self._start_application()
246
247
        # Act
248
249
        cli.run(self.path)
250
251
        # Assert
252
253
        assert self._is_application_running()
254
255
        data = self._fetch_data()
256
        assert data.status.is_running(self.application, self.computer)
257
258
    def test_case_4(self):
259
        """Verify a newly stopped local application is noted."""
260
261
        # Arrange
262
263
        self._store_data()
264
265
        status = self.data.status
266
        status.start(self.application, self.computer)
267
        self.data.status = status
268
269
        self._stop_application()
270
271
        # Act
272
273
        cli.run(self.path)
274
275
        # Assert
276
277
        assert not self._is_application_running()
278
279
        data = self._fetch_data()
280
        assert not data.status.is_running(self.application, self.computer)
281
282
    def test_case_5(self):
283
        """Verify an already stopped local application is ignored."""
284
285
        # Arrange
286
287
        self._store_data()
288
289
        self._stop_application()
290
291
        # Act
292
293
        cli.run(self.path)
294
295
        # Assert
296
297
        assert not self._is_application_running()
298
299
        data = self._fetch_data()
300
        assert not data.status.is_running(self.application, self.computer)
301