tests.test_all.TestProcesses.test_case_4() - Code Metrics - jacebrowning/mine - Measure and Improve Code Quality continuously with Scrutinizer

tests.test_all.TestProcesses.test_case_4() A
last analyzed 2020-12-17 23:19 UTC

↳ Parent: tests.test_all

Complexity

Conditions

Size

Total Lines	23
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	10
nop	1
dl	0
loc	23
rs	9.9
c	0
b	0
f	0

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

import os
import subprocess

import log
import yorm

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

from .conftest import FILES


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

jacebrowning / mine

tests.test_all.TestProcesses.test_case_4() A last analyzed 2020-12-17 23:19 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

tests.test_all.TestProcesses.test_case_4() A
last analyzed 2020-12-17 23:19 UTC