10425715883

Committed 16 Aug 2024 08:14PM UTC coverage: 67.499% (+0.05%) from 67.453%

Build # 10425715883

Build Type

Pull #1028

github

Committed by

Flix6x

Commit Message

fix: add test cases for invalid input quantities

Signed-off-by: F.N. Claessen <felix@seita.nl>

Pull Request Pull Request #1028: feature: flex context prices as quantities

Run Details

6122 of 9915 branches covered (61.74%)

Branch coverage included in aggregate %.

26 of 31 new or added lines in 4 files covered. (83.87%)

1 existing line in 1 file now uncovered.

7739 of 10620 relevant lines covered (72.87%)

4.33 hits per line

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

61.58

/flexmeasures/app.py

"""
Starting point of the Flask application.
"""

from __future__ import annotations

import time
from copy import copy
import os
from pathlib import Path
from datetime import date

from flask import Flask, g, request
from flask.cli import load_dotenv
from flask_mail import Mail
from flask_sslify import SSLify
from flask_json import FlaskJSON
from flask_cors import CORS

from redis import Redis
from rq import Queue

from flexmeasures.data.services.job_cache import JobCache


def create(  # noqa C901
    env: str | None = None,
    path_to_config: str | None = None,
    plugins: list[str] | None = None,
) -> Flask:
    """
    Create a Flask app and configure it.

    Set the environment by setting FLEXMEASURES_ENV as environment variable (also possible in .env).
    Or, overwrite any FLEXMEASURES_ENV setting by passing an env in directly (useful for testing for instance).

    A path to a config file can be passed in (otherwise a config file will be searched in the home or instance directories).

    Also, a list of plugins can be set. Usually this works as a config setting, but this is useful for automated testing.
    """

    from flexmeasures.utils import config_defaults
    from flexmeasures.utils.config_utils import read_config, configure_logging
    from flexmeasures.utils.app_utils import set_secret_key, init_sentry
    from flexmeasures.utils.error_utils import add_basic_error_handlers

    # Create app

    configure_logging()  # do this first, see https://flask.palletsprojects.com/en/2.0.x/logging
    # we're loading dotenv files manually & early (can do Flask.run(load_dotenv=False)),
    # as we need to know the ENV now (for it to be recognised by Flask()).
    load_dotenv()
    app = Flask("flexmeasures")

    if env is not None:  # overwrite
        app.config["FLEXMEASURES_ENV"] = env
    if app.config.get("FLEXMEASURES_ENV") == "testing":
        app.testing = True
    if app.config.get("FLEXMEASURES_ENV") == "development":
        app.debug = config_defaults.DevelopmentConfig.DEBUG

    # App configuration

    read_config(app, custom_path_to_config=path_to_config)
    if plugins:
        app.config["FLEXMEASURES_PLUGINS"] += plugins
    add_basic_error_handlers(app)
    if (
        app.config.get("FLEXMEASURES_ENV") not in ("development", "documentation")
        and not app.testing
    ):
        init_sentry(app)

    app.mail = Mail(app)
    FlaskJSON(app)
    CORS(app)

    # configure Redis (for redis queue)
    if app.testing:
        from fakeredis import FakeStrictRedis

        redis_conn = FakeStrictRedis(
            host="redis", port="1234"
        )  # dummy connection details
    else:
        redis_conn = Redis(
            app.config["FLEXMEASURES_REDIS_URL"],
            port=app.config["FLEXMEASURES_REDIS_PORT"],
            db=app.config["FLEXMEASURES_REDIS_DB_NR"],
            password=app.config["FLEXMEASURES_REDIS_PASSWORD"],
        )
        """ FWIW, you could use redislite like this (not on non-recent os.name=="nt" systems or PA, sadly):
            from redislite import Redis
            redis_conn = Redis("MY-DB-NAME", unix_socket_path="/tmp/my-redis.socket",
            )
        """
    app.redis_connection = redis_conn
    app.queues = dict(
        forecasting=Queue(connection=redis_conn, name="forecasting"),
        scheduling=Queue(connection=redis_conn, name="scheduling"),
        # reporting=Queue(connection=redis_conn, name="reporting"),
        # labelling=Queue(connection=redis_conn, name="labelling"),
        # alerting=Queue(connection=redis_conn, name="alerting"),
    )
    app.job_cache = JobCache(app.redis_connection)

    # Some basic security measures

    set_secret_key(app)
    if app.config.get("SECURITY_PASSWORD_SALT", None) is None:
        app.config["SECURITY_PASSWORD_SALT"] = app.config["SECRET_KEY"]
    if app.config.get("FLEXMEASURES_FORCE_HTTPS", False):
        SSLify(app)

    # Prepare profiling, if needed

    if app.config.get("FLEXMEASURES_PROFILE_REQUESTS", False):
        Path("profile_reports").mkdir(parents=True, exist_ok=True)
        try:
            import pyinstrument  # noqa F401
        except ImportError:
            app.logger.warning(
                "FLEXMEASURES_PROFILE_REQUESTS is True, but pyinstrument not installed ― I cannot produce profiling reports for requests."
            )

    # Register database and models, including user auth security handlers

    from flexmeasures.data import register_at as register_db_at

    register_db_at(app)

    # Register Reporters and Schedulers
    from flexmeasures.utils.coding_utils import get_classes_module
    from flexmeasures.data.models import reporting, planning

    reporters = get_classes_module("flexmeasures.data.models", reporting.Reporter)
    schedulers = get_classes_module("flexmeasures.data.models", planning.Scheduler)

    app.data_generators = dict()
    app.data_generators["reporter"] = copy(
        reporters
    )  # use copy to avoid mutating app.reporters
    app.data_generators["scheduler"] = schedulers

    # add auth policy

    from flexmeasures.auth import register_at as register_auth_at

    register_auth_at(app)

    # This needs to happen here because for unknown reasons, Security(app)
    # and FlaskJSON() will set this to False on their own
    if app.config.get("FLEXMEASURES_JSON_COMPACT", False) in (
        True,
        "True",
        "true",
        "1",
        "yes",
    ):
        app.json.compact = True
    else:
        app.json.compact = False

    # Register the CLI

    from flexmeasures.cli import register_at as register_cli_at

    register_cli_at(app)

    # Register the API

    from flexmeasures.api import register_at as register_api_at

    register_api_at(app)

    # Register plugins
    # If plugins register routes, they'll have precedence over standard UI
    # routes (first registration wins). However, we want to control "/" separately.

    from flexmeasures.utils.app_utils import root_dispatcher
    from flexmeasures.utils.plugin_utils import register_plugins

    app.add_url_rule("/", view_func=root_dispatcher)
    register_plugins(app)

    # Register the UI

    from flexmeasures.ui import register_at as register_ui_at

    register_ui_at(app)

    # Global template variables for both our own templates and external templates
    @app.context_processor
    def set_global_template_variables():
        return {"queue_names": app.queues.keys()}

    # Profile endpoints (if needed, e.g. during development)

    @app.before_request
    def before_request():
        if app.config.get("FLEXMEASURES_PROFILE_REQUESTS", False):
            g.start = time.time()
            try:
                import pyinstrument  # noqa F401

                g.profiler = pyinstrument.Profiler(async_mode="disabled")
                g.profiler.start()
            except ImportError:
                pass

    @app.teardown_request
    def teardown_request(exception=None):
        if app.config.get("FLEXMEASURES_PROFILE_REQUESTS", False):
            diff = time.time() - g.start
            if all([kw not in request.url for kw in ["/static", "favicon.ico"]]):
                app.logger.info(
                    f"[PROFILE] {str(round(diff, 2)).rjust(6)} seconds to serve {request.url}."
                )
                if not hasattr(g, "profiler"):
                    return app
                g.profiler.stop()
                output_html = g.profiler.output_html(timeline=True)
                endpoint = request.endpoint
                if endpoint is None:
                    endpoint = "unknown"
                today = date.today()
                profile_filename = f"pyinstrument_{endpoint}.html"
                profile_output_path = Path(
                    "profile_reports", today.strftime("%Y-%m-%d")
                )
                profile_output_path.mkdir(parents=True, exist_ok=True)
                with open(
                    os.path.join(profile_output_path, profile_filename), "w+"
                ) as f:
                    f.write(output_html)

    return app

1	"""	3✔
2	Starting point of the Flask application.
3	"""
4
5	from __future__ import annotations	6✔
6
7	import time	6✔
8	from copy import copy	6✔
9	import os	6✔
10	from pathlib import Path	6✔
11	from datetime import date	6✔
12
13	from flask import Flask, g, request	6✔
14	from flask.cli import load_dotenv	6✔
15	from flask_mail import Mail	6✔
16	from flask_sslify import SSLify	6✔
17	from flask_json import FlaskJSON	6✔
18	from flask_cors import CORS	6✔
19
20	from redis import Redis	6✔
21	from rq import Queue	6✔
22
23	from flexmeasures.data.services.job_cache import JobCache	6✔
24
25
26	def create( # noqa C901	6✔
27	env: str \| None = None,
28	path_to_config: str \| None = None,
29	plugins: list[str] \| None = None,
30	) -> Flask:
31	"""
32	Create a Flask app and configure it.
33
34	Set the environment by setting FLEXMEASURES_ENV as environment variable (also possible in .env).
35	Or, overwrite any FLEXMEASURES_ENV setting by passing an env in directly (useful for testing for instance).
36
37	A path to a config file can be passed in (otherwise a config file will be searched in the home or instance directories).
38
39	Also, a list of plugins can be set. Usually this works as a config setting, but this is useful for automated testing.
40	"""
41
42	from flexmeasures.utils import config_defaults	6✔
43	from flexmeasures.utils.config_utils import read_config, configure_logging	6✔
44	from flexmeasures.utils.app_utils import set_secret_key, init_sentry	6✔
45	from flexmeasures.utils.error_utils import add_basic_error_handlers	6✔
46
47	# Create app
48
49	configure_logging() # do this first, see https://flask.palletsprojects.com/en/2.0.x/logging	6✔
50	# we're loading dotenv files manually & early (can do Flask.run(load_dotenv=False)),
51	# as we need to know the ENV now (for it to be recognised by Flask()).
52	load_dotenv()	6✔
53	app = Flask("flexmeasures")	6✔
54
55	if env is not None: # overwrite	6✔
56	app.config["FLEXMEASURES_ENV"] = env	6✔
57	if app.config.get("FLEXMEASURES_ENV") == "testing":	6!
58	app.testing = True	6✔
59	if app.config.get("FLEXMEASURES_ENV") == "development":	6!
60	app.debug = config_defaults.DevelopmentConfig.DEBUG	×
61
62	# App configuration
63
64	read_config(app, custom_path_to_config=path_to_config)	6✔
65	if plugins:	6✔
66	app.config["FLEXMEASURES_PLUGINS"] += plugins	×
67	add_basic_error_handlers(app)	6✔
68	if (	6✔
69	app.config.get("FLEXMEASURES_ENV") not in ("development", "documentation")
70	and not app.testing
71	):
72	init_sentry(app)	×
73
74	app.mail = Mail(app)	6✔
75	FlaskJSON(app)	6✔
76	CORS(app)	6✔
77
78	# configure Redis (for redis queue)
79	if app.testing:	6✔
80	from fakeredis import FakeStrictRedis	6✔
81
82	redis_conn = FakeStrictRedis(	6✔
83	host="redis", port="1234"
84	) # dummy connection details
85	else:
86	redis_conn = Redis(	×
87	app.config["FLEXMEASURES_REDIS_URL"],
88	port=app.config["FLEXMEASURES_REDIS_PORT"],
89	db=app.config["FLEXMEASURES_REDIS_DB_NR"],
90	password=app.config["FLEXMEASURES_REDIS_PASSWORD"],
91	)
UNCOV 92	""" FWIW, you could use redislite like this (not on non-recent os.name=="nt" systems or PA, sadly):
93	from redislite import Redis
94	redis_conn = Redis("MY-DB-NAME", unix_socket_path="/tmp/my-redis.socket",
95	)
96	"""
97	app.redis_connection = redis_conn	6✔
98	app.queues = dict(	6✔
99	forecasting=Queue(connection=redis_conn, name="forecasting"),
100	scheduling=Queue(connection=redis_conn, name="scheduling"),
101	# reporting=Queue(connection=redis_conn, name="reporting"),
102	# labelling=Queue(connection=redis_conn, name="labelling"),
103	# alerting=Queue(connection=redis_conn, name="alerting"),
104	)
105	app.job_cache = JobCache(app.redis_connection)	6✔
106
107	# Some basic security measures
108
109	set_secret_key(app)	6✔
110	if app.config.get("SECURITY_PASSWORD_SALT", None) is None:	6✔
111	app.config["SECURITY_PASSWORD_SALT"] = app.config["SECRET_KEY"]	×
112	if app.config.get("FLEXMEASURES_FORCE_HTTPS", False):	6✔
113	SSLify(app)	×
114
115	# Prepare profiling, if needed
116
117	if app.config.get("FLEXMEASURES_PROFILE_REQUESTS", False):	6✔
118	Path("profile_reports").mkdir(parents=True, exist_ok=True)	×
119	try:	×
120	import pyinstrument # noqa F401	×
121	except ImportError:	×
122	app.logger.warning(	×
123	"FLEXMEASURES_PROFILE_REQUESTS is True, but pyinstrument not installed ― I cannot produce profiling reports for requests."
124	)
125
126	# Register database and models, including user auth security handlers
127
128	from flexmeasures.data import register_at as register_db_at	6✔
129
130	register_db_at(app)	6✔
131
132	# Register Reporters and Schedulers
133	from flexmeasures.utils.coding_utils import get_classes_module	6✔
134	from flexmeasures.data.models import reporting, planning	6✔
135
136	reporters = get_classes_module("flexmeasures.data.models", reporting.Reporter)	6✔
137	schedulers = get_classes_module("flexmeasures.data.models", planning.Scheduler)	6✔
138
139	app.data_generators = dict()	6✔
140	app.data_generators["reporter"] = copy(	6✔
141	reporters
142	) # use copy to avoid mutating app.reporters
143	app.data_generators["scheduler"] = schedulers	6✔
144
145	# add auth policy
146
147	from flexmeasures.auth import register_at as register_auth_at	6✔
148
149	register_auth_at(app)	6✔
150
151	# This needs to happen here because for unknown reasons, Security(app)
152	# and FlaskJSON() will set this to False on their own
153	if app.config.get("FLEXMEASURES_JSON_COMPACT", False) in (	6✔
154	True,
155	"True",
156	"true",
157	"1",
158	"yes",
159	):
160	app.json.compact = True	×
161	else:
162	app.json.compact = False	6✔
163
164	# Register the CLI
165
166	from flexmeasures.cli import register_at as register_cli_at	6✔
167
168	register_cli_at(app)	6✔
169
170	# Register the API
171
172	from flexmeasures.api import register_at as register_api_at	6✔
173
174	register_api_at(app)	6✔
175
176	# Register plugins
177	# If plugins register routes, they'll have precedence over standard UI
178	# routes (first registration wins). However, we want to control "/" separately.
179
180	from flexmeasures.utils.app_utils import root_dispatcher	6✔
181	from flexmeasures.utils.plugin_utils import register_plugins	6✔
182
183	app.add_url_rule("/", view_func=root_dispatcher)	6✔
184	register_plugins(app)	6✔
185
186	# Register the UI
187
188	from flexmeasures.ui import register_at as register_ui_at	6✔
189
190	register_ui_at(app)	6✔
191
192	# Global template variables for both our own templates and external templates
193	@app.context_processor	6✔
194	def set_global_template_variables():	6✔
195	return {"queue_names": app.queues.keys()}	6✔
196
197	# Profile endpoints (if needed, e.g. during development)
198
199	@app.before_request	6✔
200	def before_request():	6✔
201	if app.config.get("FLEXMEASURES_PROFILE_REQUESTS", False):	6✔
202	g.start = time.time()	×
203	try:	×
204	import pyinstrument # noqa F401	×
205
206	g.profiler = pyinstrument.Profiler(async_mode="disabled")	×
207	g.profiler.start()	×
208	except ImportError:	×
209	pass	×
210
211	@app.teardown_request	6✔
212	def teardown_request(exception=None):	6✔
213	if app.config.get("FLEXMEASURES_PROFILE_REQUESTS", False):	6✔
214	diff = time.time() - g.start	×
215	if all([kw not in request.url for kw in ["/static", "favicon.ico"]]):	×
216	app.logger.info(	×
217	f"[PROFILE] {str(round(diff, 2)).rjust(6)} seconds to serve {request.url}."
218	)
219	if not hasattr(g, "profiler"):	×
220	return app	×
221	g.profiler.stop()	×
222	output_html = g.profiler.output_html(timeline=True)	×
223	endpoint = request.endpoint	×
224	if endpoint is None:	×
225	endpoint = "unknown"	×
226	today = date.today()	×
227	profile_filename = f"pyinstrument_{endpoint}.html"	×
228	profile_output_path = Path(	×
229	"profile_reports", today.strftime("%Y-%m-%d")
230	)
231	profile_output_path.mkdir(parents=True, exist_ok=True)	×
232	with open(	×
233	os.path.join(profile_output_path, profile_filename), "w+"
234	) as f:
235	f.write(output_html)	×
236
237	return app	6✔

FlexMeasures / flexmeasures / 10425715883

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