# -*- coding: utf-8; -*- import datetime import decimal import os import shutil import sys import tempfile import warnings import uuid as _uuid from unittest import TestCase from unittest.mock import patch, MagicMock import pytest from mako.template import Template import wuttjamaican.enum from wuttjamaican import app as mod from wuttjamaican.exc import ConfigurationError from wuttjamaican.progress import ProgressBase from wuttjamaican.conf import WuttaConfig from wuttjamaican.util import UNSPECIFIED, get_timezone_by_name from wuttjamaican.testing import FileTestCase, ConfigTestCase from wuttjamaican.batch import BatchHandler class MockBatchHandler(BatchHandler): pass class AnotherBatchHandler(BatchHandler): pass class TestAppHandler(FileTestCase): def setUp(self): self.setup_files() self.config = WuttaConfig(appname="wuttatest") self.app = mod.AppHandler(self.config) self.config.app = self.app def test_init(self): self.assertIs(self.app.config, self.config) self.assertEqual(self.app.handlers, {}) self.assertEqual(self.app.appname, "wuttatest") def test_get_enum(self): self.assertIs(self.app.get_enum(), wuttjamaican.enum) def test_load_object(self): # just confirm the method works on a basic level; the # underlying function is tested elsewhere obj = self.app.load_object("wuttjamaican.util:UNSPECIFIED") self.assertIs(obj, UNSPECIFIED) def test_get_appdir(self): mockdir = self.mkdtemp() # default appdir with patch.object(sys, "prefix", new=mockdir): # default is returned by default appdir = self.app.get_appdir() self.assertEqual(appdir, os.path.join(mockdir, "app")) # but not if caller wants config only appdir = self.app.get_appdir(configured_only=True) self.assertIsNone(appdir) # also, cannot create if appdir path not known self.assertRaises( ValueError, self.app.get_appdir, configured_only=True, create=True ) # configured appdir self.config.setdefault("wuttatest.appdir", mockdir) appdir = self.app.get_appdir() self.assertEqual(appdir, mockdir) # appdir w/ subpath appdir = self.app.get_appdir("foo", "bar") self.assertEqual(appdir, os.path.join(mockdir, "foo", "bar")) # subpath is created self.assertEqual(len(os.listdir(mockdir)), 0) appdir = self.app.get_appdir("foo", "bar", create=True) self.assertEqual(appdir, os.path.join(mockdir, "foo", "bar")) self.assertEqual(os.listdir(mockdir), ["foo"]) self.assertEqual(os.listdir(os.path.join(mockdir, "foo")), ["bar"]) def test_make_appdir(self): # appdir is created, and 3 subfolders added by default tempdir = tempfile.mkdtemp() appdir = os.path.join(tempdir, "app") self.assertFalse(os.path.exists(appdir)) self.app.make_appdir(appdir) self.assertTrue(os.path.exists(appdir)) self.assertEqual(len(os.listdir(appdir)), 4) shutil.rmtree(tempdir) # subfolders still added if appdir already exists tempdir = tempfile.mkdtemp() self.assertTrue(os.path.exists(tempdir)) self.assertEqual(len(os.listdir(tempdir)), 0) self.app.make_appdir(tempdir) self.assertEqual(len(os.listdir(tempdir)), 4) shutil.rmtree(tempdir) def test_render_mako_template(self): output_conf = self.write_file("output.conf", "") template = Template( """\ [wutta] app_title = WuttaTest """ ) output = self.app.render_mako_template(template, {}, output_path=output_conf) self.assertEqual( output, """\ [wutta] app_title = WuttaTest """, ) with open(output_conf, "rt") as f: self.assertEqual(f.read(), output) def test_resource_path(self): result = self.app.resource_path("wuttjamaican:templates") self.assertEqual( result, os.path.join(os.path.dirname(mod.__file__), "templates") ) def test_make_session(self): try: from wuttjamaican import db except ImportError: pytest.skip("test is not relevant without sqlalchemy") session = self.app.make_session() self.assertIsInstance(session, db.Session.class_) def test_short_session(self): short_session = MagicMock() mockdb = MagicMock(short_session=short_session) with patch.dict("sys.modules", **{"wuttjamaican.db": mockdb}): with self.app.short_session(foo="bar") as s: short_session.assert_called_once_with( foo="bar", factory=self.app.make_session ) def test_get_setting(self): try: import sqlalchemy as sa from sqlalchemy import orm except ImportError: pytest.skip("test is not relevant without sqlalchemy") Session = orm.sessionmaker() engine = sa.create_engine("sqlite://") session = Session(bind=engine) session.execute( sa.text( """ create table setting ( name varchar(255) primary key, value text ); """ ) ) session.commit() value = self.app.get_setting(session, "foo") self.assertIsNone(value) session.execute(sa.text("insert into setting values ('foo', 'bar');")) value = self.app.get_setting(session, "foo") self.assertEqual(value, "bar") def test_save_setting(self): try: import sqlalchemy as sa from sqlalchemy import orm except ImportError: pytest.skip("test is not relevant without sqlalchemy") Session = orm.sessionmaker() engine = sa.create_engine("sqlite://") session = Session(bind=engine) session.execute( sa.text( """ create table setting ( name varchar(255) primary key, value text ); """ ) ) session.commit() # value null by default value = self.app.get_setting(session, "foo") self.assertIsNone(value) # unless we save a value self.app.save_setting(session, "foo", "1") session.commit() value = self.app.get_setting(session, "foo") self.assertEqual(value, "1") def test_delete_setting(self): try: import sqlalchemy as sa from sqlalchemy import orm except ImportError: pytest.skip("test is not relevant without sqlalchemy") Session = orm.sessionmaker() engine = sa.create_engine("sqlite://") session = Session(bind=engine) session.execute( sa.text( """ create table setting ( name varchar(255) primary key, value text ); """ ) ) session.commit() # value null by default value = self.app.get_setting(session, "foo") self.assertIsNone(value) # unless we save a value self.app.save_setting(session, "foo", "1") session.commit() value = self.app.get_setting(session, "foo") self.assertEqual(value, "1") # but then if we delete it, should be null again self.app.delete_setting(session, "foo") session.commit() value = self.app.get_setting(session, "foo") self.assertIsNone(value) def test_continuum_is_enabled(self): # false by default with patch.object(self.app, "providers", new={}): self.assertFalse(self.app.continuum_is_enabled()) # but "any" provider technically could enable it... class MockProvider: def continuum_is_enabled(self): return True with patch.object(self.app, "providers", new={"mock": MockProvider()}): self.assertTrue(self.app.continuum_is_enabled()) def test_model(self): try: from wuttjamaican.db import model except ImportError: pytest.skip("test not relevant without sqlalchemy") self.assertNotIn("model", self.app.__dict__) self.assertIs(self.app.model, model) def test_get_model(self): try: from wuttjamaican.db import model except ImportError: pytest.skip("test not relevant without sqlalchemy") self.assertIs(self.app.get_model(), model) def test_get_title(self): self.assertEqual(self.app.get_title(), "WuttJamaican") def test_get_node_title(self): # default self.assertEqual(self.app.get_node_title(), "WuttJamaican") # will fallback to app title self.config.setdefault("wuttatest.app_title", "WuttaTest") self.assertEqual(self.app.get_node_title(), "WuttaTest") # will read from config self.config.setdefault("wuttatest.node_title", "WuttaNode") self.assertEqual(self.app.get_node_title(), "WuttaNode") def test_get_node_type(self): # default self.assertIsNone(self.app.get_node_type()) # will read from config self.config.setdefault("wuttatest.node_type", "warehouse") self.assertEqual(self.app.get_node_type(), "warehouse") def test_get_distribution(self): try: from sqlalchemy.orm import Query except ImportError: pytest.skip("test is not relevant without sqlalchemy") # works with "non-native" objects query = Query({}) dist = self.app.get_distribution(query) self.assertEqual(dist, "SQLAlchemy") # can override dist via config self.config.setdefault("wuttatest.app_dist", "importlib_metadata") dist = self.app.get_distribution() self.assertEqual(dist, "importlib_metadata") # but the provided object takes precedence dist = self.app.get_distribution(query) self.assertEqual(dist, "SQLAlchemy") def test_get_distribution_pre_python_3_10(self): # the goal here is to get coverage for code which would only # run on python 3,9 and older, but we only need that coverage # if we are currently testing python 3.10+ if sys.version_info.major == 3 and sys.version_info.minor < 10: pytest.skip("this test is not relevant before python 3.10") importlib_metadata = MagicMock() importlib_metadata.packages_distributions = MagicMock( return_value={ "wuttjamaican": ["WuttJamaican"], "config": ["python-configuration"], } ) orig_import = __import__ def mock_import(name, *args, **kwargs): if name == "importlib.metadata": raise ImportError if name == "importlib_metadata": return importlib_metadata return orig_import(name, *args, **kwargs) with patch("builtins.__import__", side_effect=mock_import): # default should always be WuttJamaican (right..?) dist = self.app.get_distribution() self.assertEqual(dist, "WuttJamaican") # also works with "non-native" objects from config import Configuration config = Configuration({}) dist = self.app.get_distribution(config) self.assertEqual(dist, "python-configuration") # hacky sort of test, just in case we can't deduce the # package dist based on the obj - easy enough since we # have limited the packages_distributions() above dist = self.app.get_distribution(42) self.assertIsNone(dist) # can override dist via config self.config.setdefault("wuttatest.app_dist", "importlib_metadata") dist = self.app.get_distribution() self.assertEqual(dist, "importlib_metadata") # but the provided object takes precedence dist = self.app.get_distribution(config) self.assertEqual(dist, "python-configuration") # hacky test again, this time config override should win dist = self.app.get_distribution(42) self.assertEqual(dist, "importlib_metadata") def test_get_version(self): from importlib.metadata import version try: from sqlalchemy.orm import Query except ImportError: pytest.skip("test is not relevant without sqlalchemy") # works with "non-native" objects query = Query({}) ver = self.app.get_version(obj=query) self.assertEqual(ver, version("SQLAlchemy")) # random object will not yield a dist nor version ver = self.app.get_version(obj=42) self.assertIsNone(ver) # can override dist via config self.config.setdefault("wuttatest.app_dist", "python-configuration") ver = self.app.get_version() self.assertEqual(ver, version("python-configuration")) # but the provided object takes precedence ver = self.app.get_version(obj=query) self.assertEqual(ver, version("SQLAlchemy")) # can also specify the dist ver = self.app.get_version(dist="progress") self.assertEqual(ver, version("progress")) def test_make_title(self): text = self.app.make_title("foo_bar") self.assertEqual(text, "Foo Bar") def test_make_full_name(self): name = self.app.make_full_name("Fred", "", "Flintstone", "") self.assertEqual(name, "Fred Flintstone") def test_get_timezone(self): # cache is empty at first self.assertEqual(self.app.timezones, {}) # fetch default system local timezone # nb. actual value depends on machine where tests run system = self.app.get_timezone() self.assertIsInstance(system, datetime.tzinfo) # cache no longer empty self.assertEqual(len(self.app.timezones), 1) self.assertIn("default", self.app.timezones) self.assertIs(self.app.timezones["default"], system) # fetch configured default self.app.timezones.clear() # clear cache self.config.setdefault("wuttatest.timezone.default", "Africa/Addis_Ababa") default = self.app.get_timezone() self.assertIsInstance(default, datetime.tzinfo) dt = datetime.datetime(2025, 12, 16, 22, 0, tzinfo=default) self.assertEqual(default.utcoffset(dt), datetime.timedelta(hours=3)) # cache no longer empty self.assertEqual(len(self.app.timezones), 1) self.assertIn("default", self.app.timezones) self.assertIs(self.app.timezones["default"], default) # fetching again gives cached instance self.assertIs(self.app.get_timezone(), default) # fetch configured alternate self.config.setdefault("wuttatest.timezone.foo", "America/New_York") foo = self.app.get_timezone("foo") self.assertIsInstance(foo, datetime.tzinfo) self.assertIn("foo", self.app.timezones) self.assertIs(self.app.timezones["foo"], foo) # error if alternate not configured self.assertRaises(ConfigurationError, self.app.get_timezone, "bar") self.assertNotIn("bar", self.app.timezones) def test_get_timezone_name(self): # cache is empty at first self.assertEqual(self.app.timezones, {}) # fetch default system local timezone # nb. actual value depends on machine where tests run system = self.app.get_timezone_name() self.assertIsInstance(system, str) self.assertLess(0, len(system)) # cache no longer empty self.assertEqual(len(self.app.timezones), 1) self.assertIn("default", self.app.timezones) # fetch configured default self.app.timezones.clear() # clear cache self.config.setdefault("wuttatest.timezone.default", "Africa/Addis_Ababa") default = self.app.get_timezone_name() # nb. this check won't work for python 3.8 if sys.version_info >= (3, 9): self.assertEqual(default, "Africa/Addis_Ababa") # cache no longer empty self.assertEqual(len(self.app.timezones), 1) self.assertIn("default", self.app.timezones) # fetch configured alternate self.config.setdefault("wuttatest.timezone.foo", "America/New_York") foo = self.app.get_timezone_name("foo") # nb. this check won't work for python 3.8 if sys.version_info >= (3, 9): self.assertEqual(foo, "America/New_York") self.assertIn("foo", self.app.timezones) # error if alternate not configured self.assertRaises(ConfigurationError, self.app.get_timezone_name, "bar") self.assertNotIn("bar", self.app.timezones) def test_localtime(self): dt = self.app.localtime() self.assertIsInstance(dt, datetime.datetime) self.assertIsNotNone(dt.tzinfo) def test_make_utc(self): dt = self.app.make_utc() self.assertIsInstance(dt, datetime.datetime) self.assertIsNone(dt.tzinfo) def test_make_str_uuid(self): uuid = self.app.make_str_uuid() self.assertEqual(len(uuid), 32) def test_make_true_uuid(self): uuid = self.app.make_true_uuid() self.assertIsInstance(uuid, _uuid.UUID) def test_make_uuid(self): # TODO: temporary behavior uuid = self.app.make_uuid() self.assertEqual(len(uuid), 32) def test_progress_loop(self): def act(obj, i): pass # with progress self.app.progress_loop(act, [1, 2, 3], ProgressBase, message="whatever") # without progress self.app.progress_loop(act, [1, 2, 3], None, message="whatever") def test_get_session(self): try: import sqlalchemy as sa from sqlalchemy import orm except ImportError: pytest.skip("test not relevant without sqlalchemy") model = self.app.model user = model.User() self.assertIsNone(self.app.get_session(user)) Session = orm.sessionmaker() engine = sa.create_engine("sqlite://") mysession = Session(bind=engine) mysession.add(user) session = self.app.get_session(user) self.assertIs(session, mysession) def test_render_boolean(self): # null self.assertEqual(self.app.render_boolean(None), "") # true self.assertEqual(self.app.render_boolean(True), "Yes") # false self.assertEqual(self.app.render_boolean(False), "No") def test_render_currency(self): # null self.assertEqual(self.app.render_currency(None), "") # basic decimal example value = decimal.Decimal("42.00") self.assertEqual(self.app.render_currency(value), "$42.00") # basic float example value = 42.00 self.assertEqual(self.app.render_currency(value), "$42.00") # decimal places will be rounded value = decimal.Decimal("42.12345") self.assertEqual(self.app.render_currency(value), "$42.12") # but we can declare the scale value = decimal.Decimal("42.12345") self.assertEqual(self.app.render_currency(value, scale=4), "$42.1234") # negative numbers get parens value = decimal.Decimal("-42.42") self.assertEqual(self.app.render_currency(value), "($42.42)") def test_render_date(self): self.assertEqual(self.app.render_date(None), "") dt = datetime.date(2024, 12, 11) self.assertEqual(self.app.render_date(dt), "2024-12-11") def test_render_datetime(self): self.config.setdefault("wuttatest.timezone.default", "America/Los_Angeles") tzlocal = get_timezone_by_name("America/Los_Angeles") # null value self.assertEqual(self.app.render_datetime(None), "") # naive UTC dt = datetime.datetime(2024, 12, 17, 1, 12) self.assertEqual( self.app.render_datetime(dt, local=True), "2024-12-16 17:12-0800" ) self.assertEqual(self.app.render_datetime(dt, local=False), "2024-12-17 01:12") # aware UTC dt = datetime.datetime(2024, 12, 17, 1, 12, tzinfo=datetime.timezone.utc) self.assertEqual( self.app.render_datetime(dt, local=True), "2024-12-16 17:12-0800" ) self.assertEqual( self.app.render_datetime(dt, local=False), "2024-12-17 01:12+0000" ) # aware local dt = datetime.datetime(2024, 12, 16, 19, 12, tzinfo=tzlocal) self.assertEqual( self.app.render_datetime(dt, local=True), "2024-12-16 19:12-0800" ) self.assertEqual( self.app.render_datetime(dt, local=False), "2024-12-16 19:12-0800" ) # as html dt = datetime.datetime(2024, 12, 16, 19, 12, tzinfo=tzlocal) html = self.app.render_datetime(dt, local=True, html=True) self.assertTrue(html.startswith('