build.rsudp.test.make_test_settings() - Code Metrics - Inspection of "Update and rename CONTRIBUTORS.md to CONTRIBUTORS...." - raspishake/rsudp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( e96c3d...44474a )

by Ian

created 2020-05-26 20:36 UTC

build.rsudp.test.make_test_settings() A

↳ Parent: build.rsudp.test

Complexity

Conditions

Size

Total Lines	65
Code Lines	22

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	22
dl	0
loc	65
rs	9.352
c	0
b	0
f	0
cc	2
nop	2

How to fix Long Method

import os, sys
from rsudp import COLOR, printM, printW, printE
from queue import Queue
import socket
import json
import time
import pkg_resources as pr

SENDER = 'test.py'
TEST = {
	# permissions
	'p_log_dir':			['log directory               ', False],
	'p_log_std':			['stdout logging              ', False],
	'p_log_file':			['logging to file             ', False],
	'p_output_dirs':		['output directory structure  ', False],
	'p_screenshot_dir':		['screenshot directory        ', False],
	'p_data_dir':			['data directory              ', False],

	# network
	'n_port':				['port                        ', False],
	'n_internet':			['internet                    ', False],
	'n_inventory':			['inventory fetch             ', False],

	# dependencies
	'd_pydub':				['pydub dependencies          ', False],
	'd_matplotlib':			['matplotlib backend          ', False],

	# core
	'c_data':				['receiving data              ', False],
	'c_processing':			['processing data             ', False],
	'c_ALARM':				['ALARM message               ', False],
	'c_RESET':				['RESET message               ', False],
	'c_IMGPATH':			['IMGPATH message             ', False],
	'c_TERM':				['TERM message                ', False],
}

TRANS = {
	True: COLOR['green'] + 'PASS' + COLOR['white'],
	False: COLOR['red'] + 'FAIL' + COLOR['white']
}

PORT = 18888

def make_test_settings(settings, inet=False):
	'''
	Get the default settings and return settings for testing.

	The default settings are modified in the following way:

	======================================== ===================
	Setting                                  Value
	======================================== ===================
	 ``settings['settings']['station']``      ``'R24FA'``
	 ``settings['alert']['threshold']``       ``2``
	 ``settings['alert']['reset']``           ``0.5``
	 ``settings['alert']['lowpass']``         ``9``
	 ``settings['alert']['highpass']``        ``0.8``
	 ``settings['plot']['channels']``         ``['all']``
	 ``settings['plot']['duration']``         ``60``
	 ``settings['plot']['deconvolve']``       ``True``
	 ``settings['plot']['units']``            ``'CHAN'``
	 ``settings['plot']['eq_screenshots']``   ``True``
	 ``settings['alertsound']['enabled']``    ``True``
	 ``settings['rsam']['enabled']``          ``True``
	 ``settings['rsam']['debug']``            ``True``
	======================================== ===================

	.. note::

		If there is no internet connection detected, the station
		name will default to ``'Z0000'`` so that no time is wasted
		trying to download an inventory from the Raspberry Shake
		FDSN service.

	:param dict settings: settings dictionary (will be modified from :ref:`defaults`)
	:param bool inet: whether or not the internet test passed
	:rtype: dict
	:return: settings dictionary to test with
	'''
	settings = json.loads(settings)

	settings['settings']['port'] = PORT
	if inet:
		settings['settings']['station'] = 'R24FA'
	else:
		settings['settings']['station'] = 'Z0000'


	settings['alert']['threshold'] = 2
	settings['alert']['reset'] = 0.5
	settings['alert']['lowpass'] = 9
	settings['alert']['highpass'] = 0.8

	settings['plot']['channels'] = ['all']
	settings['plot']['duration'] = 60
	settings['plot']['deconvolve'] = True
	settings['plot']['units'] = 'CHAN'
	settings['plot']['eq_screenshots'] = True

	settings['alertsound']['enabled'] = True

	settings['rsam']['enabled'] = True
	settings['rsam']['debug'] = True
	settings['rsam']['interval'] = 5
	settings['rsam']['fwaddr'] = "127.0.0.1"
	settings['rsam']['fwport'] = 4444

	return settings


def cancel_tests(settings, MPL, plot, quiet):
	'''
	Cancel some tests if they don't need to be run.

	:param dict settings: the dictionary of settings for program execution
	:param bool plot: whether or not to plot (``False`` == no plot)
	:param bool quiet: whether or not to play sounds (``True`` == no sound)
	:rtype: dict
	:return: settings dictionary to test with
	'''
	global TEST

	if plot:
		if MPL:
			TEST['d_matplotlib'][1] = True
		else:
			printW('matplotlib backend failed to load')
	else:
		settings['plot']['enabled'] = False
		del TEST['d_matplotlib']
		del TEST['c_IMGPATH']
		printM('Plot is disabled')

	if quiet:
		settings['alertsound']['enabled'] = False
		del TEST['d_pydub']
		printM('Alert sound is disabled')
	return settings


def permissions(dp):
	'''
	Test write permissions for the specified directory.

	:param str dp: the directory path to test permissions for
	:rtype: bool
	:return: if ``True``, the test was successful, ``False`` otherwise
	'''
	dp = os.path.join(dp, 'test')
	try:
		with open(dp, 'w') as f:
			f.write('testing\n')
		os.remove(dp)
		return True
	except Exception as e:
		printE(e)
		return False

def datadir_permissions(testdir):
	'''
	Test write permissions in the data directory (``./data`` by default)

	:param str testdir: The directory to test permissions for
	:rtype: bool
	:return: the output of :py:func:`rsudp.test.permissions`

	'''
	return permissions('%s/data/' % testdir)

def ss_permissions(testdir):
	'''
	Test write permissions in the screenshots directory (``./screenshots`` by default)

	:param str testdir: The directory to test permissions for
	:rtype: bool
	:return: the output of :py:func:`rsudp.test.permissions`

	'''
	return permissions('%s/screenshots/' % testdir)

def logdir_permissions(logdir='/tmp/rsudp'):
	'''
	Test write permissions in the log directory (``/tmp/rsudp`` by default)

	:param str logdir: The log directory to test permissions for
	:rtype: bool
	:return: the output of :py:func:`rsudp.test.permissions`
	'''
	return permissions(logdir)

def is_connected(hostname):
	'''
	Test for an internet connection. 

	:param str hostname: The hostname to test with
	:rtype: bool
	:return: ``True`` if connection is successful, ``False`` otherwise
	'''
	try:
		# see if we can resolve the host name -- tells us if there is
		# a DNS listening
		host = socket.gethostbyname(hostname)
		# connect to the host -- tells us if the host is actually
		# reachable
		s = socket.create_connection((host, 80), 2)
		s.close()
		return True
	except:
		pass
	return False



1			import os, sys
2			from rsudp import COLOR, printM, printW, printE
3			from queue import Queue
4			import socket
5			import json
6			import time
7			import pkg_resources as pr
8
9			SENDER = 'test.py'
10			TEST = {
11			# permissions
12			'p_log_dir': ['log directory ', False],
13			'p_log_std': ['stdout logging ', False],
14			'p_log_file': ['logging to file ', False],
15			'p_output_dirs': ['output directory structure ', False],
16			'p_screenshot_dir': ['screenshot directory ', False],
17			'p_data_dir': ['data directory ', False],
18
19			# network
20			'n_port': ['port ', False],
21			'n_internet': ['internet ', False],
22			'n_inventory': ['inventory fetch ', False],
23
24			# dependencies
25			'd_pydub': ['pydub dependencies ', False],
26			'd_matplotlib': ['matplotlib backend ', False],
27
28			# core
29			'c_data': ['receiving data ', False],
30			'c_processing': ['processing data ', False],
31			'c_ALARM': ['ALARM message ', False],
32			'c_RESET': ['RESET message ', False],
33			'c_IMGPATH': ['IMGPATH message ', False],
34			'c_TERM': ['TERM message ', False],
35			}
36
37			TRANS = {
38			True: COLOR['green'] + 'PASS' + COLOR['white'],
39			False: COLOR['red'] + 'FAIL' + COLOR['white']
40			}
41
42			PORT = 18888
43
44			def make_test_settings(settings, inet=False):
45			'''
46			Get the default settings and return settings for testing.
47
48			The default settings are modified in the following way:
49
50			======================================== ===================
51			Setting Value
52			======================================== ===================
53			``settings['settings']['station']`` ``'R24FA'``
54			``settings['alert']['threshold']`` ``2``
55			``settings['alert']['reset']`` ``0.5``
56			``settings['alert']['lowpass']`` ``9``
57			``settings['alert']['highpass']`` ``0.8``
58			``settings['plot']['channels']`` ``['all']``
59			``settings['plot']['duration']`` ``60``
60			``settings['plot']['deconvolve']`` ``True``
61			``settings['plot']['units']`` ``'CHAN'``
62			``settings['plot']['eq_screenshots']`` ``True``
63			``settings['alertsound']['enabled']`` ``True``
64			``settings['rsam']['enabled']`` ``True``
65			``settings['rsam']['debug']`` ``True``
66			======================================== ===================
67
68			.. note::
69
70			If there is no internet connection detected, the station
71			name will default to ``'Z0000'`` so that no time is wasted
72			trying to download an inventory from the Raspberry Shake
73			FDSN service.
74
75			:param dict settings: settings dictionary (will be modified from :ref:`defaults`)
76			:param bool inet: whether or not the internet test passed
77			:rtype: dict
78			:return: settings dictionary to test with
79			'''
80			settings = json.loads(settings)
81
82			settings['settings']['port'] = PORT
83			if inet:
84			settings['settings']['station'] = 'R24FA'
85			else:
86			settings['settings']['station'] = 'Z0000'
87
88
89			settings['alert']['threshold'] = 2
90			settings['alert']['reset'] = 0.5
91			settings['alert']['lowpass'] = 9
92			settings['alert']['highpass'] = 0.8
93
94			settings['plot']['channels'] = ['all']
95			settings['plot']['duration'] = 60
96			settings['plot']['deconvolve'] = True
97			settings['plot']['units'] = 'CHAN'
98			settings['plot']['eq_screenshots'] = True
99
100			settings['alertsound']['enabled'] = True
101
102			settings['rsam']['enabled'] = True
103			settings['rsam']['debug'] = True
104			settings['rsam']['interval'] = 5
105			settings['rsam']['fwaddr'] = "127.0.0.1"
106			settings['rsam']['fwport'] = 4444
107
108			return settings
109
110
111			def cancel_tests(settings, MPL, plot, quiet):
112			'''
113			Cancel some tests if they don't need to be run.
114
115			:param dict settings: the dictionary of settings for program execution
116			:param bool plot: whether or not to plot (``False`` == no plot)
117			:param bool quiet: whether or not to play sounds (``True`` == no sound)
118			:rtype: dict
119			:return: settings dictionary to test with
120			'''
121			global TEST
122
123			if plot:
124			if MPL:
125			TEST['d_matplotlib'][1] = True
126			else:
127			printW('matplotlib backend failed to load')
128			else:
129			settings['plot']['enabled'] = False
130			del TEST['d_matplotlib']
131			del TEST['c_IMGPATH']
132			printM('Plot is disabled')
133
134			if quiet:
135			settings['alertsound']['enabled'] = False
136			del TEST['d_pydub']
137			printM('Alert sound is disabled')
138			return settings
139
140
141			def permissions(dp):
142			'''
143			Test write permissions for the specified directory.
144
145			:param str dp: the directory path to test permissions for
146			:rtype: bool
147			:return: if ``True``, the test was successful, ``False`` otherwise
148			'''
149			dp = os.path.join(dp, 'test')
150			try:
151			with open(dp, 'w') as f:
152			f.write('testing\n')
153			os.remove(dp)
154			return True
155			except Exception as e:
156			printE(e)
157			return False
158
159			def datadir_permissions(testdir):
160			'''
161			Test write permissions in the data directory (``./data`` by default)
162
163			:param str testdir: The directory to test permissions for
164			:rtype: bool
165			:return: the output of :py:func:`rsudp.test.permissions`
166
167			'''
168			return permissions('%s/data/' % testdir)
169
170			def ss_permissions(testdir):
171			'''
172			Test write permissions in the screenshots directory (``./screenshots`` by default)
173
174			:param str testdir: The directory to test permissions for
175			:rtype: bool
176			:return: the output of :py:func:`rsudp.test.permissions`
177
178			'''
179			return permissions('%s/screenshots/' % testdir)
180
181			def logdir_permissions(logdir='/tmp/rsudp'):
182			'''
183			Test write permissions in the log directory (``/tmp/rsudp`` by default)
184
185			:param str logdir: The log directory to test permissions for
186			:rtype: bool
187			:return: the output of :py:func:`rsudp.test.permissions`
188			'''
189			return permissions(logdir)
190
191			def is_connected(hostname):
192			'''
193			Test for an internet connection.
194
195			:param str hostname: The hostname to test with
196			:rtype: bool
197			:return: ``True`` if connection is successful, ``False`` otherwise
198			'''
199			try:
200			# see if we can resolve the host name -- tells us if there is
201			# a DNS listening
202			host = socket.gethostbyname(hostname)
203			# connect to the host -- tells us if the host is actually
204			# reachable
205			s = socket.create_connection((host, 80), 2)
206			s.close()
207			return True
208			except:
209			pass
210			return False
211
212

raspishake / rsudp

Push — master ( e96c3d...44474a )

build.rsudp.test.make_test_settings() A

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like