TestProcesses.teardown_method() - Code Metrics - Inspection of "Release v1.4" - jacebrowning/mine - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#79)

by Jace

created 2017-04-18 18:06 UTC

TestProcesses.teardown_method() A

↳ Parent: TestProcesses

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	2
c	2
b	0
f	0
dl	0
loc	5
rs	9.4285

# pylint: disable=misplaced-comparison-constant,no-self-use

import os
import subprocess
import logging

import yorm

from mine import cli
from mine.models import (Application, Computer, Data, ProgramConfig,
                         State, Status, ProgramStatus)

from .conftest import FILES


log = logging.getLogger(__name__)


class TestFiles:
    """Integration tests for creating files."""

    def test_data(self):
        """Verify a sample file is created."""
        path = os.path.join(FILES, 'mine.yml')

        if os.path.exists(path):
            os.remove(path)

        data = Data()
        yorm.sync(data, path)

        itunes = Application('itunes')
        itunes.versions.mac = ''
        itunes.versions.windows = 'iTunes.exe'

        iphoto = Application('iphoto')
        iphoto.versions.mac = 'iPhoto'

        mac = Computer('macbook', 'Other.local')
        mac2 = Computer('macbook-pro')

        configuration = ProgramConfig()
        configuration.applications = [itunes, iphoto]
        configuration.computers = [mac, mac2]

        data.config = configuration

        mac_state = State('macbook-pro')
        mac_state.timestamp.started = 444

        itunes_status = Status('itunes')
        itunes_status.computers = [mac_state]

        status = ProgramStatus()
        status.applications = [itunes_status]
        status.counter = 499

        data.status = status

        assert os.path.exists(path)

    def test_data_out(self):
        """Verify a sample file is created."""
        path = os.path.join(FILES, 'mine-out.yml')

        if os.path.exists(path):
            os.remove(path)

        data = Data()
        yorm.sync(data, path)

        itunes = Application('itunes')
        itunes.versions.mac = ''
        itunes.versions.windows = 'iTunes.exe'

        iphoto = Application('iphoto')
        iphoto.versions.mac = 'iPhoto'

        mac = Computer('macbook', 'Jaces-MacBook', 'AA:BB:CC:DD:EE:FF')
        mac2 = Computer('macbook-pro', 'Jaces-MacBook-2', '11:22:33:44:55:66')

        configuration = ProgramConfig()
        configuration.applications = [itunes, iphoto]
        configuration.computers = [mac, mac2]

        data.config = configuration

        mac_state = State('macbook-pro')
        mac_state.timestamp.started = 444

        itunes_status = Status('itunes')
        itunes_status.computers = [mac_state]

        status = ProgramStatus()
        status.applications = [itunes_status]
        status.counter = 499

        data.status = status

        assert os.path.exists(path)

    def test_data_in(self):
        """Verify a sample file is loaded."""
        path = os.path.join(FILES, 'mine-in.yml')

        data = Data()
        yorm.sync(data, path)

        assert data.config.applications
        for application in data.config.applications:
            if application.name == 'slack':
                break
        else:
            assert False


class TestProcesses:
    """Integration tests for tracking and stopping processes."""

    NAME = "example application"

    application = Application(NAME)
    application.versions.linux = 'yes'
    application.versions.mac = 'yes'
    application.versions.windows = 'yes.exe'

    computer = None
    data = None

    path = os.path.join(FILES, 'mine.tmp.yml')

    _process = None

    def _store_data(self):
        """Set up initial data file for tests."""
        self.data = Data()
        self.data.config.applications.append(self.application)
        self.computer = self.data.config.computers.get_current()  # pylint: disable=no-member
        yorm.sync(self.data, self.path)

    def _fetch_data(self):
        """Read the final data file back for verification."""
        data = Data()
        yorm.sync(data, self.path)
        return data

    def _start_application(self):
        """Start the example application."""
        if not self._process:
            # TODO: get filename from the application object
            self._process = subprocess.Popen(['yes'], stdout=subprocess.PIPE)
        log.info("%s is started", self.application)

    def _stop_application(self):
        """Stop the example application."""
        if self._process:
            if self._process.poll() is None:
                self._process.kill()
            self._process = None
        log.info("%s is stopped", self.application)

    def _is_application_running(self):
        """Determine if the sample application is running."""
        return self._process and self._process.poll() is None

    def teardown_method(self, _):
        """Stop the sample application and clean up the file."""
        self._stop_application()
        if os.path.exists(self.path):
            os.remove(self.path)

    def test_case_1(self):
        """Verify a newly running remote application takes over."""

        # Arrange

        self._store_data()

        # start the application and manually mark it as running
        self._start_application()
        status = self.data.status
        status.start(self.application, self.computer)
        self.data.status = status
        assert self.data.status.is_running(self.application, self.computer)

        # manually mark the application as running on a remote computer
        computer = Computer('other', 'Other.local', 'AA:BB:CC:DD:EE:FF')
        status = self.data.status
        status.start(self.application, computer)
        self.data.status = status
        assert self.data.status.is_running(self.application, computer)

        # Act

        cli.run(self.path, cleanup=False)

        # Assert

        # verify the application is no longer running
        assert not self._is_application_running()
        # verify the application is marked as running remotely
        data = self._fetch_data()
        assert not data.status.is_running(self.application, self.computer)
        assert data.status.is_running(self.application, computer)

    def test_case_2(self):
        """Verify a newly running local application takes over."""

        # Arrange

        self._store_data()

        # manually mark the application as running on a remote computer
        computer = Computer('other', 'Other.local', 'AA:BB:CC:DD:EE:FF')
        status = self.data.status
        status.start(self.application, computer)
        self.data.status = status
        assert self.data.status.is_running(self.application, computer)

        # start the application
        self._start_application()

        # Act
        cli.run(self.path, cleanup=False)

        # Assert

        assert self._is_application_running()

        data = self._fetch_data()
        assert data.status.is_running(self.application, self.computer)
        assert data.status.is_running(self.application, computer)

    def test_case_3(self):
        """Verify an already running local application is ignored."""

        # Arrange

        self._store_data()

        status = self.data.status
        status.start(self.application, self.computer)
        self.data.status = status

        self._start_application()

        # Act

        cli.run(self.path)

        # Assert

        assert self._is_application_running()

        data = self._fetch_data()
        assert data.status.is_running(self.application, self.computer)

    def test_case_4(self):
        """Verify a newly stopped local application is noted."""

        # Arrange

        self._store_data()

        status = self.data.status
        status.start(self.application, self.computer)
        self.data.status = status

        self._stop_application()

        # Act

        cli.run(self.path)

        # Assert

        assert not self._is_application_running()

        data = self._fetch_data()
        assert not data.status.is_running(self.application, self.computer)

    def test_case_6(self):
        """Verify an already stopped local application is ignored."""

        # Arrange

        self._store_data()

        self._stop_application()

        # Act

        cli.run(self.path)

        # Assert

        assert not self._is_application_running()

        data = self._fetch_data()
        assert not data.status.is_running(self.application, self.computer)


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			def test_data(self):
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			def test_data_out(self):
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_6(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

jacebrowning / mine

Pull Request — master (#79)

TestProcesses.teardown_method() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like