13800528386

Committed 11 Mar 2025 11:55PM UTC coverage: 99.991% (-0.009%) from 100.0%

Build # 13800528386

Build Type

Pull #1158

github

Committed by

web-flow

Commit Message

Merge f9853ebba into 1130e3a64

Pull Request Pull Request #1158: Quotas bug fix

Run Details

25 of 25 new or added lines in 4 files covered. (100.0%)

1 existing line in 1 file now uncovered.

11232 of 11233 relevant lines covered (99.99%)

1.0 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.4

/tethys_apps/harvester.py

"""
********************************************************************************
* Name: app_harvester.py
* Author: Nathan Swain and Scott Christensen
* Created On: August 19, 2013
* Copyright: (c) Brigham Young University 2013
* License: BSD 2-Clause
********************************************************************************
"""

import inspect
import logging
import pkgutil
from sqlite3 import ProgrammingError as SqliteProgrammingError
from django.db.utils import ProgrammingError
from django.core.exceptions import ObjectDoesNotExist
from tethys_apps.base import TethysAppBase, TethysExtensionBase
from tethys_apps.base.testing.environment import is_testing_environment

tethys_log = logging.getLogger("tethys." + __name__)


class SingletonHarvester:
    """
    Collects information for initiating apps
    """

    extensions = []
    extension_modules = {}
    app_modules = {}
    apps = []
    _instance = None
    BLUE = "\033[94m"
    GREEN = "\033[92m"
    WARNING = "\033[93m"
    FAIL = "\033[91m"
    ENDC = "\033[0m"

    def harvest(self):
        """
        Harvest apps and extensions.
        """
        self.harvest_extensions()
        self.harvest_apps()

    def harvest_extensions(self):
        """
        Searches for and loads Tethys extensions.
        """
        try:
            if not is_testing_environment():
                print(self.BLUE + "Loading Tethys Extensions..." + self.ENDC)

            import tethysext

            tethys_extensions = dict()
            for _, modname, ispkg in pkgutil.iter_modules(tethysext.__path__):
                if ispkg:
                    tethys_extensions[modname] = "tethysext.{}".format(modname)
            self._harvest_extension_instances(tethys_extensions)
        except Exception:
            """DO NOTHING"""

    def harvest_apps(self):
        """
        Searches the apps package for apps
        """
        # Notify user harvesting is taking place

        try:
            if not is_testing_environment():
                print(self.BLUE + "Loading Tethys Apps..." + self.ENDC)

            import tethysapp

            tethys_apps = dict()
            for _, modname, ispkg in pkgutil.iter_modules(tethysapp.__path__):
                if ispkg:
                    tethys_apps[modname] = "tethysapp.{}".format(modname)

            # Harvest App Instances
            self._harvest_app_instances(tethys_apps)

        except Exception:
            """DO NOTHING"""

    def get_url_patterns(self, url_namespaces=None):
        """
        Generate the url pattern lists for each app and namespace them accordingly.
        """
        app_url_patterns = dict()
        ext_url_patterns = dict()
        ws_url_patterns = dict()
        apps = self.apps
        if url_namespaces:
            apps = [app for app in apps if app.url_namespace in url_namespaces]

        for app in apps:
            app_url_patterns.update(app.url_patterns["http"])

        for app in apps:
            ws_url_patterns.update(app.url_patterns["websocket"])

        for extension in self.extensions:
            ext_url_patterns.update(extension.url_patterns["http"])

        url_patterns = {
            "app_url_patterns": app_url_patterns,
            "ext_url_patterns": ext_url_patterns,
            "ws_url_patterns": ws_url_patterns,
        }

        return url_patterns

    def get_handler_patterns(self, url_namespaces=None):
        """
        Generate the url handler pattern lists for each app and namespace them accordingly.
        """
        http_handler_patterns = dict()
        ws_handler_patterns = dict()
        apps = self.apps
        if url_namespaces:
            apps = [app for app in apps if app.url_namespace in url_namespaces]

        for app in apps:
            http_handler_patterns.update(app.handler_patterns["http"])

        for app in apps:
            ws_handler_patterns.update(app.handler_patterns["websocket"])

        handler_patterns = {
            "http_handler_patterns": http_handler_patterns,
            "ws_handler_patterns": ws_handler_patterns,
        }

        return handler_patterns

    def __new__(cls):
        """
        Make App Harvester a Singleton
        """
        if not cls._instance:
            cls._instance = super().__new__(cls)

        return cls._instance

    @staticmethod
    def _validate_extension(extension):
        """
        Validate the given extension.
        Args:
            extension(module_obj): ext module object of the Tethys extension.

        Returns:
            module_obj or None: returns validated module object or None if not valid.
        """
        return extension

    @staticmethod
    def _validate_app(app):
        """
        Validate the app data that needs to be validated. Returns either the app if valid or None if not valid.
        """
        # Remove prepended slash if included
        if app.icon != "" and app.icon[0] == "/":
            app.icon = app.icon[1:]

        # Validate color
        if app.color != "" and app.color[0] != "#":
            # Add hash
            app.color = "#{0}".format(app.color)

        # Must be 6 or 3 digit hex color (7 or 4 with hash symbol)
        if len(app.color) != 7 and len(app.color) != 4:
            app.color = ""

        return app

    def _harvest_extension_instances(self, extension_packages):
        """
        Locate the extension class, instantiate it, and save for later use.

        Arg:
            extension_packages(dict<name, extension_package>): Dictionary where keys are the name of the extension and value is the extension package module object.
        """  # noqa:E501
        valid_ext_instances = []
        valid_extension_modules = {}
        loaded_extensions = []

        for extension_name, extension_package in extension_packages.items():
            try:
                # Import the "ext" module from the extension package
                ext_module = __import__(extension_package + ".ext", fromlist=[""])

                # Retrieve the members of the ext_module and iterate through
                # them to find the the class that inherits from TethysExtensionBase.
                for name, obj in inspect.getmembers(ext_module):
                    try:
                        # issubclass() will fail if obj is not a class
                        if (issubclass(obj, TethysExtensionBase)) and (
                            obj is not TethysExtensionBase
                        ):
                            # Assign a handle to the class
                            ExtensionClass = getattr(ext_module, name)

                            # Instantiate app and validate
                            ext_instance = ExtensionClass()
                            validated_ext_instance = self._validate_extension(
                                ext_instance
                            )

                            # sync app with Tethys db
                            ext_instance.sync_with_tethys_db()

                            # compile valid apps
                            if validated_ext_instance:
                                valid_ext_instances.append(validated_ext_instance)
                                valid_extension_modules[extension_name] = (
                                    extension_package
                                )

                                # Notify user that the app has been loaded
                                loaded_extensions.append(extension_name)

                            # We found the extension class so we're done
                            break

                    except TypeError:
                        continue
            except Exception:
                tethys_log.exception(
                    "Extension {0} not loaded because of the following error:".format(
                        extension_package
                    )
                )
                continue

        # Save valid apps
        self.extensions = valid_ext_instances
        self.extension_modules = valid_extension_modules

        # Update user
        if not is_testing_environment():
            print(
                self.BLUE
                + "Tethys Extensions Loaded: "
                + self.ENDC
                + "{0}".format(", ".join(loaded_extensions))
                + "\n"
            )

    def _harvest_app_instances(self, app_packages_list):
        """
        Search each app package for the app.py module. Find the AppBase class in the app.py
        module and instantiate it. Save the list of instantiated AppBase classes.
        """
        valid_app_instance_list = []
        valid_app_modules = {}
        loaded_apps = []
        for app_name, app_package in app_packages_list.items():
            # Skip these things
            if app_package in [
                "__init__.py",
                "__init__.pyc",
                ".gitignore",
                ".DS_Store",
            ]:
                continue

            try:
                # Import the app.py module from the custom app package programmatically
                # (e.g.: apps.apps.<custom_package>.app)

                app_module = __import__(app_package + ".app", fromlist=[""])

                for name, obj in inspect.getmembers(app_module):
                    # Retrieve the members of the app_module and iterate through
                    # them to find the class that inherits from AppBase.
                    try:
                        # issubclass() will fail if obj is not a class
                        if (issubclass(obj, TethysAppBase)) and (
                            obj is not TethysAppBase
                        ):
                            # Assign a handle to the class
                            AppClass = getattr(app_module, name)

                            # Instantiate app and validate
                            app_instance = AppClass()
                            validated_app_instance = self._validate_app(app_instance)

                            # sync app with Tethys db
                            app_instance.sync_with_tethys_db()

                            # load/validate app url patterns
                            try:
                                app_instance.url_patterns
                            except Exception:
                                tethys_log.exception(
                                    "App {0} not loaded because of an issue with loading urls:".format(
                                        app_package
                                    )
                                )
                                app_instance.remove_from_db()
                                continue

                            # load/validate app handler patterns
                            try:
                                app_instance.handler_patterns
                            except Exception:
                                tethys_log.exception(
                                    "App {0} not loaded because of an issue with loading handlers:".format(
                                        app_package
                                    )
                                )
                                app_instance.remove_from_db()
                                continue

                            # register app permissions
                            try:
                                app_instance.register_app_permissions()
                            except ProgrammingError:
                                tethys_log.warning(
                                    "Unable to register app permissions. django_content_type "
                                    "table does not exist"
                                )
                            except SqliteProgrammingError:
                                tethys_log.warning(
                                    "Unable to register app permissions. django_content_type "
                                    "table does not exist in sqlite3"
                                )
                            except ObjectDoesNotExist as e:
                                tethys_log.warning(e)

                            # compile valid apps
                            if validated_app_instance:
                                valid_app_modules[app_name] = app_package
                                valid_app_instance_list.append(validated_app_instance)

                                # Notify user that the app has been loaded
                                loaded_apps.append(app_name)

                            # We found the app class so we're done
                            break

                    except TypeError:
                        continue
            except Exception:
                tethys_log.exception(
                    "App {0} not loaded because of the following error:".format(
                        app_package
                    )
                )
                continue

        # Save valid apps
        self.apps = valid_app_instance_list
        self.app_modules = valid_app_modules

        # Update user
        if not is_testing_environment():
            print(
                self.BLUE
                + "Tethys Apps Loaded: "
                + self.ENDC
                + "{0}".format(", ".join(loaded_apps))
                + "\n"
            )

1	"""
2	********************************************************************************
3	* Name: app_harvester.py
4	* Author: Nathan Swain and Scott Christensen
5	* Created On: August 19, 2013
6	* Copyright: (c) Brigham Young University 2013
7	* License: BSD 2-Clause
8	********************************************************************************
9	"""
10
11	import inspect	1✔
12	import logging	1✔
13	import pkgutil	1✔
14	from sqlite3 import ProgrammingError as SqliteProgrammingError	1✔
15	from django.db.utils import ProgrammingError	1✔
16	from django.core.exceptions import ObjectDoesNotExist	1✔
17	from tethys_apps.base import TethysAppBase, TethysExtensionBase	1✔
18	from tethys_apps.base.testing.environment import is_testing_environment	1✔
19
20	tethys_log = logging.getLogger("tethys." + __name__)	1✔
21
22
23	class SingletonHarvester:	1✔
24	"""
25	Collects information for initiating apps
26	"""
27
28	extensions = []	1✔
29	extension_modules = {}	1✔
30	app_modules = {}	1✔
31	apps = []	1✔
32	_instance = None	1✔
33	BLUE = "\033[94m"	1✔
34	GREEN = "\033[92m"	1✔
35	WARNING = "\033[93m"	1✔
36	FAIL = "\033[91m"	1✔
37	ENDC = "\033[0m"	1✔
38
39	def harvest(self):	1✔
40	"""
41	Harvest apps and extensions.
42	"""
43	self.harvest_extensions()	1✔
44	self.harvest_apps()	1✔
45
46	def harvest_extensions(self):	1✔
47	"""
48	Searches for and loads Tethys extensions.
49	"""
50	try:	1✔
51	if not is_testing_environment():	1✔
52	print(self.BLUE + "Loading Tethys Extensions..." + self.ENDC)	1✔
53
54	import tethysext	1✔
55
56	tethys_extensions = dict()	1✔
57	for _, modname, ispkg in pkgutil.iter_modules(tethysext.__path__):	1✔
58	if ispkg:	1✔
59	tethys_extensions[modname] = "tethysext.{}".format(modname)	1✔
60	self._harvest_extension_instances(tethys_extensions)	1✔
61	except Exception:	1✔
62	"""DO NOTHING"""	1✔
63
64	def harvest_apps(self):	1✔
65	"""
66	Searches the apps package for apps
67	"""
68	# Notify user harvesting is taking place
69
70	try:	1✔
71	if not is_testing_environment():	1✔
72	print(self.BLUE + "Loading Tethys Apps..." + self.ENDC)	1✔
73
74	import tethysapp	1✔
75
76	tethys_apps = dict()	1✔
77	for _, modname, ispkg in pkgutil.iter_modules(tethysapp.__path__):	1✔
78	if ispkg:	1✔
79	tethys_apps[modname] = "tethysapp.{}".format(modname)	1✔
80
81	# Harvest App Instances
82	self._harvest_app_instances(tethys_apps)	1✔
83
84	except Exception:	1✔
85	"""DO NOTHING"""	1✔
86
87	def get_url_patterns(self, url_namespaces=None):	1✔
88	"""
89	Generate the url pattern lists for each app and namespace them accordingly.
90	"""
91	app_url_patterns = dict()	1✔
92	ext_url_patterns = dict()	1✔
93	ws_url_patterns = dict()	1✔
94	apps = self.apps	1✔
95	if url_namespaces:	1✔
96	apps = [app for app in apps if app.url_namespace in url_namespaces]	1✔
97
98	for app in apps:	1✔
99	app_url_patterns.update(app.url_patterns["http"])	1✔
100
101	for app in apps:	1✔
102	ws_url_patterns.update(app.url_patterns["websocket"])	1✔
103
104	for extension in self.extensions:	1✔
105	ext_url_patterns.update(extension.url_patterns["http"])	1✔
106
107	url_patterns = {	1✔
108	"app_url_patterns": app_url_patterns,
109	"ext_url_patterns": ext_url_patterns,
110	"ws_url_patterns": ws_url_patterns,
111	}
112
113	return url_patterns	1✔
114
115	def get_handler_patterns(self, url_namespaces=None):	1✔
116	"""
117	Generate the url handler pattern lists for each app and namespace them accordingly.
118	"""
119	http_handler_patterns = dict()	1✔
120	ws_handler_patterns = dict()	1✔
121	apps = self.apps	1✔
122	if url_namespaces:	1✔
123	apps = [app for app in apps if app.url_namespace in url_namespaces]	1✔
124
125	for app in apps:	1✔
126	http_handler_patterns.update(app.handler_patterns["http"])	1✔
127
128	for app in apps:	1✔
129	ws_handler_patterns.update(app.handler_patterns["websocket"])	1✔
130
131	handler_patterns = {	1✔
132	"http_handler_patterns": http_handler_patterns,
133	"ws_handler_patterns": ws_handler_patterns,
134	}
135
136	return handler_patterns	1✔
137
138	def __new__(cls):	1✔
139	"""
140	Make App Harvester a Singleton
141	"""
142	if not cls._instance:	1✔
143	cls._instance = super().__new__(cls)	1✔
144
145	return cls._instance	1✔
146
147	@staticmethod	1✔
148	def _validate_extension(extension):	1✔
149	"""
150	Validate the given extension.
151	Args:
152	extension(module_obj): ext module object of the Tethys extension.
153
154	Returns:
155	module_obj or None: returns validated module object or None if not valid.
156	"""
157	return extension	1✔
158
159	@staticmethod	1✔
160	def _validate_app(app):	1✔
161	"""
162	Validate the app data that needs to be validated. Returns either the app if valid or None if not valid.
163	"""
164	# Remove prepended slash if included
165	if app.icon != "" and app.icon[0] == "/":	1✔
166	app.icon = app.icon[1:]	1✔
167
168	# Validate color
169	if app.color != "" and app.color[0] != "#":	1✔
170	# Add hash
171	app.color = "#{0}".format(app.color)	1✔
172
173	# Must be 6 or 3 digit hex color (7 or 4 with hash symbol)
174	if len(app.color) != 7 and len(app.color) != 4:	1✔
175	app.color = ""	1✔
176
177	return app	1✔
178
179	def _harvest_extension_instances(self, extension_packages):	1✔
180	"""
181	Locate the extension class, instantiate it, and save for later use.
182
183	Arg:
184	extension_packages(dict<name, extension_package>): Dictionary where keys are the name of the extension and value is the extension package module object.
185	""" # noqa:E501
186	valid_ext_instances = []	1✔
187	valid_extension_modules = {}	1✔
188	loaded_extensions = []	1✔
189
190	for extension_name, extension_package in extension_packages.items():	1✔
191	try:	1✔
192	# Import the "ext" module from the extension package
193	ext_module = __import__(extension_package + ".ext", fromlist=[""])	1✔
194
195	# Retrieve the members of the ext_module and iterate through
196	# them to find the the class that inherits from TethysExtensionBase.
197	for name, obj in inspect.getmembers(ext_module):	1✔
198	try:	1✔
199	# issubclass() will fail if obj is not a class
200	if (issubclass(obj, TethysExtensionBase)) and (	1✔
201	obj is not TethysExtensionBase
202	):
203	# Assign a handle to the class
204	ExtensionClass = getattr(ext_module, name)	1✔
205
206	# Instantiate app and validate
207	ext_instance = ExtensionClass()	1✔
208	validated_ext_instance = self._validate_extension(	1✔
209	ext_instance
210	)
211
212	# sync app with Tethys db
213	ext_instance.sync_with_tethys_db()	1✔
214
215	# compile valid apps
216	if validated_ext_instance:	1✔
217	valid_ext_instances.append(validated_ext_instance)	1✔
218	valid_extension_modules[extension_name] = (	1✔
219	extension_package
220	)
221
222	# Notify user that the app has been loaded
223	loaded_extensions.append(extension_name)	1✔
224
225	# We found the extension class so we're done
226	break	1✔
227
228	except TypeError:	1✔
229	continue	1✔
230	except Exception:	1✔
231	tethys_log.exception(	1✔
232	"Extension {0} not loaded because of the following error:".format(
233	extension_package
234	)
235	)
236	continue	1✔
237
238	# Save valid apps
239	self.extensions = valid_ext_instances	1✔
240	self.extension_modules = valid_extension_modules	1✔
241
242	# Update user
243	if not is_testing_environment():	1✔
244	print(	1✔
245	self.BLUE
246	+ "Tethys Extensions Loaded: "
247	+ self.ENDC
248	+ "{0}".format(", ".join(loaded_extensions))
249	+ "\n"
250	)
251
252	def _harvest_app_instances(self, app_packages_list):	1✔
253	"""
254	Search each app package for the app.py module. Find the AppBase class in the app.py
255	module and instantiate it. Save the list of instantiated AppBase classes.
256	"""
257	valid_app_instance_list = []	1✔
258	valid_app_modules = {}	1✔
259	loaded_apps = []	1✔
260	for app_name, app_package in app_packages_list.items():	1✔
261	# Skip these things
262	if app_package in [	1✔
263	"__init__.py",
264	"__init__.pyc",
265	".gitignore",
266	".DS_Store",
267	]:
268	continue	1✔
269
270	try:	1✔
271	# Import the app.py module from the custom app package programmatically
272	# (e.g.: apps.apps.<custom_package>.app)
273
274	app_module = __import__(app_package + ".app", fromlist=[""])	1✔
275
276	for name, obj in inspect.getmembers(app_module):	1✔
277	# Retrieve the members of the app_module and iterate through
278	# them to find the class that inherits from AppBase.
279	try:	1✔
280	# issubclass() will fail if obj is not a class
281	if (issubclass(obj, TethysAppBase)) and (	1✔
282	obj is not TethysAppBase
283	):
284	# Assign a handle to the class
285	AppClass = getattr(app_module, name)	1✔
286
287	# Instantiate app and validate
288	app_instance = AppClass()	1✔
289	validated_app_instance = self._validate_app(app_instance)	1✔
290
291	# sync app with Tethys db
292	app_instance.sync_with_tethys_db()	1✔
293
294	# load/validate app url patterns
295	try:	1✔
296	app_instance.url_patterns	1✔
297	except Exception:	1✔
298	tethys_log.exception(	1✔
299	"App {0} not loaded because of an issue with loading urls:".format(
300	app_package
301	)
302	)
303	app_instance.remove_from_db()	1✔
304	continue	1✔
305
306	# load/validate app handler patterns
307	try:	1✔
308	app_instance.handler_patterns	1✔
309	except Exception:	1✔
310	tethys_log.exception(	1✔
311	"App {0} not loaded because of an issue with loading handlers:".format(
312	app_package
313	)
314	)
315	app_instance.remove_from_db()	1✔
316	continue	1✔
317
318	# register app permissions
319	try:	1✔
320	app_instance.register_app_permissions()	1✔
321	except ProgrammingError:	1✔
322	tethys_log.warning(	1✔
323	"Unable to register app permissions. django_content_type "
324	"table does not exist"
325	)
326	except SqliteProgrammingError:	1✔
UNCOV 327	tethys_log.warning(	×
328	"Unable to register app permissions. django_content_type "
329	"table does not exist in sqlite3"
330	)
331	except ObjectDoesNotExist as e:	1✔
332	tethys_log.warning(e)	1✔
333
334	# compile valid apps
335	if validated_app_instance:	1✔
336	valid_app_modules[app_name] = app_package	1✔
337	valid_app_instance_list.append(validated_app_instance)	1✔
338
339	# Notify user that the app has been loaded
340	loaded_apps.append(app_name)	1✔
341
342	# We found the app class so we're done
343	break	1✔
344
345	except TypeError:	1✔
346	continue	1✔
347	except Exception:	1✔
348	tethys_log.exception(	1✔
349	"App {0} not loaded because of the following error:".format(
350	app_package
351	)
352	)
353	continue	1✔
354
355	# Save valid apps
356	self.apps = valid_app_instance_list	1✔
357	self.app_modules = valid_app_modules	1✔
358
359	# Update user
360	if not is_testing_environment():	1✔
361	print(	1✔
362	self.BLUE
363	+ "Tethys Apps Loaded: "
364	+ self.ENDC
365	+ "{0}".format(", ".join(loaded_apps))
366	+ "\n"
367	)

tethysplatform / tethys / 13800528386

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous