DbusTest.test_dbus() - Code Metrics - coala-analyzer/coala - Measure and Improve Code Quality continuously with Scrutinizer

DbusTest.test_dbus() B
last analyzed 2016-09-16 10:08 UTC

↳ Parent: DbusTest

Complexity

Conditions

Size

Total Lines

105

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
dl	0
loc	105
rs	8.2857
c	0
b	0
f	0

How to fix Long Method

import os
import subprocess
import sys
import time
import unittest
from unittest.case import SkipTest

from coalib.misc import Constants

try:
    import dbus
    # Needed to determine if test needs skipping
    from gi.repository import GLib

except ImportError as err:
    raise SkipTest('python-dbus or python-gi is not installed')


def make_test_server():
    # Make a dbus service in a new process. It cannot be in this process
    # as that gives SegmentationFaults because the same bus is being used.

    # For some reason this also fails on some systems if moved to another file
    return subprocess.Popen([
        sys.executable,
        '-c',
        """
import sys
import dbus
import dbus.mainloop.glib
from gi.repository import GLib
from coalib.output.dbus.DbusServer import DbusServer
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
print('Creating session bus ...')
session_bus = dbus.SessionBus()
dbus_name = dbus.service.BusName("org.coala_analyzer.v1.test", session_bus)
print('Creating DbbusServer object ...')
dbus_server = DbusServer(session_bus, "/org/coala_analyzer/v1/test",
                         on_disconnected=lambda: GLib.idle_add(sys.exit))
mainloop = GLib.MainLoop()
print('Starting GLib mainloop ...')
mainloop.run()
"""],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE)


class DbusTest(unittest.TestCase):

    def setUp(self):
        self.config_path = os.path.abspath(
            os.path.join(os.path.dirname(__file__),
                         "dbus_test_files",
                         ".coafile"))
        self.testcode_c_path = os.path.abspath(
            os.path.join(os.path.dirname(__file__),
                         "dbus_test_files",
                         "testcode.c"))

        self.subprocess = make_test_server()
        trials_left = 50

        while trials_left > 0:
            time.sleep(0.1)
            trials_left = trials_left - 1
            try:
                self.connect_to_test_server()
                continue
            except dbus.exceptions.DBusException as exception:
                if trials_left == 0:
                    print("Stdout:")
                    print(self.subprocess.stdout.read().decode("utf-8"))
                    print("Stderr:")
                    print(self.subprocess.stderr.read().decode("utf-8"))
                    raise exception

    def connect_to_test_server(self):
        self.bus = dbus.SessionBus()
        self.remote_object = self.bus.get_object("org.coala_analyzer.v1.test",
                                                 "/org/coala_analyzer/v1/test")

    def test_dbus(self):
        self.document_object_path = self.remote_object.CreateDocument(
            self.testcode_c_path,
            dbus_interface="org.coala_analyzer.v1")

        self.assertRegex(str(self.document_object_path),
                         r"^/org/coala_analyzer/v1/test/\d+/documents/\d+$")

        self.document_object = self.bus.get_object(
            "org.coala_analyzer.v1.test",
            self.document_object_path)

        config_file = self.document_object.SetConfigFile(
            "dummy_config",
            dbus_interface="org.coala_analyzer.v1")
        self.assertEqual(config_file, "dummy_config")

        config_file = self.document_object.GetConfigFile(
            dbus_interface="org.coala_analyzer.v1")
        self.assertEqual(config_file, "dummy_config")

        config_file = self.document_object.FindConfigFile(
            dbus_interface="org.coala_analyzer.v1")
        self.assertEqual(config_file, self.config_path)

        analysis = self.document_object.Analyze(
            dbus_interface="org.coala_analyzer.v1")

        self.maxDiff = None
        print(analysis)

        # Run some basic analysis with good debug messages.
        self.assertEqual(analysis[0], 1, "Exit code was not 1.")
        self.assertEqual(len(analysis[1]), 0, "Unexpected log messages found.")

        sections = analysis[2]
        self.assertEqual(len(sections), 1, "Expected only 1 section to run.")

        section = sections[0]
        self.assertEqual(section[0], "default",
                         "Expected section to be named 'default'.")
        self.assertTrue(section[1], "Section did not execute successfully.")
        self.assertEqual(len(section[2]), 2, "Expected 2 results in section.")

        # Remove the ids as they are hashes and cannot be asserted.
        for result in section[2]:
            result['id'] = 0

        # We also test as a dictionary as dbus should be able to convert
        # it into the correct python types.
        self.assertEqual(analysis,
                         (1,
                          [],
                          [('default',
                            True,
                            [{'debug_msg': '',
                              'additional_info': '',
                              'file': '',
                              'id': 0,
                              'line_nr': "",
                              'message': 'test msg',
                              'origin': 'LocalTestBear',
                              'severity': 'NORMAL',
                              'confidence': '100'},
                             {'debug_msg': '',
                              'additional_info': '',
                              'file': self.testcode_c_path,
                              'id': 0,
                              'line_nr': "",
                              'message': 'test msg',
                              'origin': 'GlobalTestBear',
                              'severity': 'NORMAL',
                              'confidence': '100'}])]))

        config_file = self.document_object.SetConfigFile(
            self.config_path + "2",
            dbus_interface="org.coala_analyzer.v1")
        analysis = self.document_object.Analyze(
            dbus_interface="org.coala_analyzer.v1")
        self.assertEqual(analysis[0], 255)
        self.assertEqual(analysis[1][1]["log_level"], "ERROR")
        self.assertEqual(analysis[1][1]["message"], Constants.CRASH_MESSAGE)

        # Skip file if file pattern doesn't match
        # Also test if 2 documents can be opened simultaneously
        self.document_object_path = self.remote_object.CreateDocument(
            "test.unknown_ext",
            dbus_interface="org.coala_analyzer.v1")
        self.document_object = self.bus.get_object(
            "org.coala_analyzer.v1.test",
            self.document_object_path)
        config_file = self.document_object.SetConfigFile(
            self.config_path,
            dbus_interface="org.coala_analyzer.v1")
        analysis = self.document_object.Analyze(
            dbus_interface="org.coala_analyzer.v1")
        self.assertEqual(analysis, (0, [], []))

        self.remote_object.DisposeDocument(
            self.testcode_c_path,
            dbus_interface="org.coala_analyzer.v1")

        self.remote_object.DisposeDocument(
            "test.unknown_ext",
            dbus_interface="org.coala_analyzer.v1")

    def tearDown(self):
        if self.subprocess:
            self.subprocess.kill()


1			import os
2			import subprocess
3			import sys
4			import time
5			import unittest
6			from unittest.case import SkipTest
7
8			from coalib.misc import Constants
9
10			try:
11			import dbus
12			# Needed to determine if test needs skipping
13			from gi.repository import GLib
			0 ignored issues – show Unused Code introduced 2016-02-22 07:01 UTC by Report Bug Copy Issue Report Unused GLib imported from gi.repository Loading history...
14			except ImportError as err:
15			raise SkipTest('python-dbus or python-gi is not installed')
16
17
18			def make_test_server():
19			# Make a dbus service in a new process. It cannot be in this process
20			# as that gives SegmentationFaults because the same bus is being used.
21
22			# For some reason this also fails on some systems if moved to another file
23			return subprocess.Popen([
24			sys.executable,
25			'-c',
26			"""
27			import sys
28			import dbus
29			import dbus.mainloop.glib
30			from gi.repository import GLib
31			from coalib.output.dbus.DbusServer import DbusServer
32			dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
33			print('Creating session bus ...')
34			session_bus = dbus.SessionBus()
35			dbus_name = dbus.service.BusName("org.coala_analyzer.v1.test", session_bus)
36			print('Creating DbbusServer object ...')
37			dbus_server = DbusServer(session_bus, "/org/coala_analyzer/v1/test",
38			on_disconnected=lambda: GLib.idle_add(sys.exit))
39			mainloop = GLib.MainLoop()
40			print('Starting GLib mainloop ...')
41			mainloop.run()
42			"""],
43			stdout=subprocess.PIPE,
44			stderr=subprocess.PIPE)
45
46
47			class DbusTest(unittest.TestCase):
48
49			def setUp(self):
50			self.config_path = os.path.abspath(
51			os.path.join(os.path.dirname(__file__),
52			"dbus_test_files",
53			".coafile"))
54			self.testcode_c_path = os.path.abspath(
55			os.path.join(os.path.dirname(__file__),
56			"dbus_test_files",
57			"testcode.c"))
58
59			self.subprocess = make_test_server()
60			trials_left = 50
61
62			while trials_left > 0:
63			time.sleep(0.1)
64			trials_left = trials_left - 1
65			try:
66			self.connect_to_test_server()
67			continue
68			except dbus.exceptions.DBusException as exception:
69			if trials_left == 0:
70			print("Stdout:")
71			print(self.subprocess.stdout.read().decode("utf-8"))
72			print("Stderr:")
73			print(self.subprocess.stderr.read().decode("utf-8"))
74			raise exception
75
76			def connect_to_test_server(self):
77			self.bus = dbus.SessionBus()
78			self.remote_object = self.bus.get_object("org.coala_analyzer.v1.test",
79			"/org/coala_analyzer/v1/test")
80
81			def test_dbus(self):
82			self.document_object_path = self.remote_object.CreateDocument(
83			self.testcode_c_path,
84			dbus_interface="org.coala_analyzer.v1")
85
86			self.assertRegex(str(self.document_object_path),
87			r"^/org/coala_analyzer/v1/test/\d+/documents/\d+$")
88
89			self.document_object = self.bus.get_object(
90			"org.coala_analyzer.v1.test",
91			self.document_object_path)
92
93			config_file = self.document_object.SetConfigFile(
94			"dummy_config",
95			dbus_interface="org.coala_analyzer.v1")
96			self.assertEqual(config_file, "dummy_config")
97
98			config_file = self.document_object.GetConfigFile(
99			dbus_interface="org.coala_analyzer.v1")
100			self.assertEqual(config_file, "dummy_config")
101
102			config_file = self.document_object.FindConfigFile(
103			dbus_interface="org.coala_analyzer.v1")
104			self.assertEqual(config_file, self.config_path)
105
106			analysis = self.document_object.Analyze(
107			dbus_interface="org.coala_analyzer.v1")
108
109			self.maxDiff = None
110			print(analysis)
111
112			# Run some basic analysis with good debug messages.
113			self.assertEqual(analysis[0], 1, "Exit code was not 1.")
114			self.assertEqual(len(analysis[1]), 0, "Unexpected log messages found.")
115
116			sections = analysis[2]
117			self.assertEqual(len(sections), 1, "Expected only 1 section to run.")
118
119			section = sections[0]
120			self.assertEqual(section[0], "default",
121			"Expected section to be named 'default'.")
122			self.assertTrue(section[1], "Section did not execute successfully.")
123			self.assertEqual(len(section[2]), 2, "Expected 2 results in section.")
124
125			# Remove the ids as they are hashes and cannot be asserted.
126			for result in section[2]:
127			result['id'] = 0
128
129			# We also test as a dictionary as dbus should be able to convert
130			# it into the correct python types.
131			self.assertEqual(analysis,
132			(1,
133			[],
134			[('default',
135			True,
136			[{'debug_msg': '',
137			'additional_info': '',
138			'file': '',
139			'id': 0,
140			'line_nr': "",
141			'message': 'test msg',
142			'origin': 'LocalTestBear',
143			'severity': 'NORMAL',
144			'confidence': '100'},
145			{'debug_msg': '',
146			'additional_info': '',
147			'file': self.testcode_c_path,
148			'id': 0,
149			'line_nr': "",
150			'message': 'test msg',
151			'origin': 'GlobalTestBear',
152			'severity': 'NORMAL',
153			'confidence': '100'}])]))
154
155			config_file = self.document_object.SetConfigFile(
156			self.config_path + "2",
157			dbus_interface="org.coala_analyzer.v1")
158			analysis = self.document_object.Analyze(
159			dbus_interface="org.coala_analyzer.v1")
160			self.assertEqual(analysis[0], 255)
161			self.assertEqual(analysis[1][1]["log_level"], "ERROR")
162			self.assertEqual(analysis[1][1]["message"], Constants.CRASH_MESSAGE)
163
164			# Skip file if file pattern doesn't match
165			# Also test if 2 documents can be opened simultaneously
166			self.document_object_path = self.remote_object.CreateDocument(
167			"test.unknown_ext",
168			dbus_interface="org.coala_analyzer.v1")
169			self.document_object = self.bus.get_object(
170			"org.coala_analyzer.v1.test",
171			self.document_object_path)
172			config_file = self.document_object.SetConfigFile(
173			self.config_path,
174			dbus_interface="org.coala_analyzer.v1")
175			analysis = self.document_object.Analyze(
176			dbus_interface="org.coala_analyzer.v1")
177			self.assertEqual(analysis, (0, [], []))
178
179			self.remote_object.DisposeDocument(
180			self.testcode_c_path,
181			dbus_interface="org.coala_analyzer.v1")
182
183			self.remote_object.DisposeDocument(
184			"test.unknown_ext",
185			dbus_interface="org.coala_analyzer.v1")
186
187			def tearDown(self):
188			if self.subprocess:
189			self.subprocess.kill()
190

coala-analyzer / coala

DbusTest.test_dbus() B last analyzed 2016-09-16 10:08 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like

DbusTest.test_dbus() B
last analyzed 2016-09-16 10:08 UTC