23385248069

Committed 21 Mar 2026 05:50PM UTC coverage: 73.202% (-2.5%) from 75.69%

Build # 23385248069

Build Type

push

github

Committed by

RustyBower

Commit Message

Fix test to match DB fallback behavior for missing output files

The endpoint now returns check.output from DB (200) instead of 404
when the on-disk file doesn't exist.

Coverage Stats

3726 of 5090 relevant lines covered (73.2%)

0.73 hits per line

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

93.9

/scoring_engine/models/setting.py

import json
import logging

from sqlalchemy import Column, Integer, Text, desc

from scoring_engine.db import db
from scoring_engine.models.base import Base

logger = logging.getLogger(__name__)

CACHE_PREFIX = "setting:"
CACHE_TTL = 60


def _get_redis():
    """Return a Redis client using the application config.

    Returns None when Redis is unavailable (e.g. tests running with
    cache_type=null or no Redis server).
    """
    try:
        import redis

        from scoring_engine.config import config

        if config.cache_type != "redis":
            return None
        return redis.Redis(
            host=config.redis_host,
            port=config.redis_port,
            password=config.redis_password or None,
            decode_responses=True,
        )
    except Exception:
        return None


class Setting(Base):
    __tablename__ = "settings"
    id = Column(Integer, primary_key=True)
    name = Column(Text, nullable=False)
    _value_text = Column(Text, nullable=False)
    _value_type = Column(Text, nullable=False)

    def __init__(self, *args, **kwargs):
        self.name = kwargs["name"]
        self._value_text = str(kwargs["value"])
        self.value = kwargs["value"]
        super(Base, self).__init__()

    def map_value_type(self, value):
        if type(value) is bool:
            return "Boolean"
        else:
            return "String"

    def convert_value_type(self):
        if self._value_type == "Boolean":
            if self._value_text == "False":
                return False
            else:
                return True
        else:
            return self._value_text

    @property
    def value(self):
        return self.convert_value_type()

    @value.setter
    def value(self, value):
        self._value_type = self.map_value_type(value)
        self._value_text = str(value)

    @classmethod
    def get_setting(cls, name):
        """Get a setting by name with Redis caching.

        Checks Redis first (shared across all workers). On miss, queries the
        database and populates the Redis cache with a 60-second TTL.

        When Redis is unavailable the method falls back to a direct DB query.
        """
        # Try Redis cache
        r = _get_redis()
        if r is not None:
            try:
                cached = r.get(CACHE_PREFIX + name)
                if cached is not None:
                    data = json.loads(cached)
                    # Build a transient Setting without hitting the DB
                    setting = cls.__new__(cls)
                    setting.id = data["id"]
                    setting.name = name
                    setting._value_text = data["value_text"]
                    setting._value_type = data["value_type"]
                    # Merge into the current session so callers can modify + commit
                    return db.session.merge(setting, load=False)
            except Exception:
                logger.debug("Redis cache read failed for setting %s", name, exc_info=True)

        # Cache miss — query DB.
        setting = db.session.query(Setting).filter(Setting.name == name).order_by(desc(Setting.id)).first()
        if setting and r is not None:
            try:
                payload = json.dumps(
                    {
                        "id": setting.id,
                        "value_text": setting._value_text,
                        "value_type": setting._value_type,
                    }
                )
                r.set(CACHE_PREFIX + name, payload, ex=CACHE_TTL)
            except Exception:
                logger.debug("Redis cache write failed for setting %s", name, exc_info=True)
        return setting

    @classmethod
    def clear_cache(cls, name=None):
        """Clear the settings cache in Redis.

        If name is provided, only clear that specific setting.
        Otherwise, clear all cached settings.
        """
        r = _get_redis()
        if r is None:
            return
        try:
            if name:
                r.delete(CACHE_PREFIX + name)
            else:
                keys = r.keys(CACHE_PREFIX + "*")
                if keys:
                    r.delete(*keys)
        except Exception:
            logger.debug("Redis cache clear failed", exc_info=True)

1	import json	1✔
2	import logging	1✔
3
4	from sqlalchemy import Column, Integer, Text, desc	1✔
5
6	from scoring_engine.db import db	1✔
7	from scoring_engine.models.base import Base	1✔
8
9	logger = logging.getLogger(__name__)	1✔
10
11	CACHE_PREFIX = "setting:"	1✔
12	CACHE_TTL = 60	1✔
13
14
15	def _get_redis():	1✔
16	"""Return a Redis client using the application config.
17
18	Returns None when Redis is unavailable (e.g. tests running with
19	cache_type=null or no Redis server).
20	"""
21	try:	1✔
22	import redis	1✔
23
24	from scoring_engine.config import config	1✔
25
26	if config.cache_type != "redis":	1✔
27	return None	1✔
28	return redis.Redis(	×
29	host=config.redis_host,
30	port=config.redis_port,
31	password=config.redis_password or None,
32	decode_responses=True,
33	)
34	except Exception:	1✔
35	return None	1✔
36
37
38	class Setting(Base):	1✔
39	__tablename__ = "settings"	1✔
40	id = Column(Integer, primary_key=True)	1✔
41	name = Column(Text, nullable=False)	1✔
42	_value_text = Column(Text, nullable=False)	1✔
43	_value_type = Column(Text, nullable=False)	1✔
44
45	def __init__(self, args, *kwargs):	1✔
46	self.name = kwargs["name"]	1✔
47	self._value_text = str(kwargs["value"])	1✔
48	self.value = kwargs["value"]	1✔
49	super(Base, self).__init__()	1✔
50
51	def map_value_type(self, value):	1✔
52	if type(value) is bool:	1✔
53	return "Boolean"	1✔
54	else:
55	return "String"	1✔
56
57	def convert_value_type(self):	1✔
58	if self._value_type == "Boolean":	1✔
59	if self._value_text == "False":	1✔
60	return False	1✔
61	else:
62	return True	1✔
63	else:
64	return self._value_text	1✔
65
66	@property	1✔
67	def value(self):	1✔
68	return self.convert_value_type()	1✔
69
70	@value.setter	1✔
71	def value(self, value):	1✔
72	self._value_type = self.map_value_type(value)	1✔
73	self._value_text = str(value)	1✔
74
75	@classmethod	1✔
76	def get_setting(cls, name):	1✔
77	"""Get a setting by name with Redis caching.
78
79	Checks Redis first (shared across all workers). On miss, queries the
80	database and populates the Redis cache with a 60-second TTL.
81
82	When Redis is unavailable the method falls back to a direct DB query.
83	"""
84	# Try Redis cache
85	r = _get_redis()	1✔
86	if r is not None:	1✔
87	try:	1✔
88	cached = r.get(CACHE_PREFIX + name)	1✔
89	if cached is not None:	1✔
90	data = json.loads(cached)	1✔
91	# Build a transient Setting without hitting the DB
92	setting = cls.__new__(cls)	1✔
93	setting.id = data["id"]	1✔
94	setting.name = name	×
95	setting._value_text = data["value_text"]	×
96	setting._value_type = data["value_type"]	×
97	# Merge into the current session so callers can modify + commit
98	return db.session.merge(setting, load=False)	×
99	except Exception:	1✔
100	logger.debug("Redis cache read failed for setting %s", name, exc_info=True)	1✔
101
102	# Cache miss — query DB.
103	setting = db.session.query(Setting).filter(Setting.name == name).order_by(desc(Setting.id)).first()	1✔
104	if setting and r is not None:	1✔
105	try:	1✔
106	payload = json.dumps(	1✔
107	{
108	"id": setting.id,
109	"value_text": setting._value_text,
110	"value_type": setting._value_type,
111	}
112	)
113	r.set(CACHE_PREFIX + name, payload, ex=CACHE_TTL)	1✔
114	except Exception:	1✔
115	logger.debug("Redis cache write failed for setting %s", name, exc_info=True)	1✔
116	return setting	1✔
117
118	@classmethod	1✔
119	def clear_cache(cls, name=None):	1✔
120	"""Clear the settings cache in Redis.
121
122	If name is provided, only clear that specific setting.
123	Otherwise, clear all cached settings.
124	"""
125	r = _get_redis()	1✔
126	if r is None:	1✔
127	return	1✔
128	try:	1✔
129	if name:	1✔
130	r.delete(CACHE_PREFIX + name)	1✔
131	else:
132	keys = r.keys(CACHE_PREFIX + "*")	1✔
133	if keys:	1✔
134	r.delete(*keys)	1✔
135	except Exception:	1✔
136	logger.debug("Redis cache clear failed", exc_info=True)	1✔

scoringengine / scoringengine / 23385248069

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