feat(iot): Sub 7 — per-sensor polling interval + rate-limit + entech seed
Per-sensor override on fp.tank.sensor.poll_interval_minutes with a
company-wide default (res.company.x_fc_default_poll_interval_minutes,
default 30) exposed in Settings → Fusion Plating → IoT. Single
lookup helper _fp_effective_poll_interval_minutes keeps downstream
call sites simple. Read-only poll_interval_display Char ("30 min
(default)" / "15 min (override)") keeps units unambiguous per user
request.
Ingest endpoint /fp/iot/ingest drops readings that arrive inside a
sensor's effective interval, returning {accepted, skipped} so the Pi
agent can log it. Pi-side interval stays its own concern.
Post-init hook seeds 5 small tanks + 20 big tanks (10 active, 10
inactive) with 1 temperature + 1 pH sensor each → 25 tanks, 50
sensors. Idempotent (keyed by tank.code, with_context(active_test=
False)). Opt-in via ir.config_parameter
fusion_plating_iot.seed_entech_tanks = '1' so a fresh install
elsewhere doesn't auto-seed. Flag set on entech today; 27 tanks / 52
sensors now live (2 pilot + 25 seeded).
Smoke on entech: 14/14 assertions pass including idempotency and
rate-limit conditions.
fusion_plating_iot → 19.0.2.0.0
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsinghpal
@@ -5,7 +5,7 @@
|
||||
|
||||
{
|
||||
'name': 'Fusion Plating — IoT Integration',
|
||||
'version': '19.0.1.0.0',
|
||||
'version': '19.0.2.0.0',
|
||||
'category': 'Manufacturing/Plating',
|
||||
'summary': 'Wire physical tank sensors to Fusion Plating — live '
|
||||
'temperature / chemistry readings with auto quality holds '
|
||||
@@ -51,6 +51,7 @@ Part of the Fusion Plating product family by Nexa Systems Inc.
|
||||
'views/fp_tank_sensor_views.xml',
|
||||
'views/fp_tank_reading_views.xml',
|
||||
'views/fusion_plating_tank_views.xml',
|
||||
'views/res_config_settings_views.xml',
|
||||
'views/fp_iot_menu.xml',
|
||||
],
|
||||
'post_init_hook': 'post_init_hook',
|
||||
|
||||
@@ -31,7 +31,7 @@ Returns 200 + `{ok: true, accepted: N}` on success, 401 on auth fail,
|
||||
import hmac
|
||||
import json
|
||||
import logging
|
||||
from datetime import datetime, timezone
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from odoo import http
|
||||
from odoo.http import request, Response
|
||||
@@ -109,6 +109,7 @@ class FpIotIngestController(http.Controller):
|
||||
Sensor = request.env['fp.tank.sensor'].sudo()
|
||||
Reading = request.env['fp.tank.reading'].sudo()
|
||||
accepted = 0
|
||||
skipped_interval = 0
|
||||
unknown_serials = []
|
||||
for r in readings:
|
||||
serial = (r.get('device_serial') or '').strip()
|
||||
@@ -122,6 +123,23 @@ class FpIotIngestController(http.Controller):
|
||||
value = float(r.get('value'))
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
|
||||
# Sub 7 — per-sensor rate-limit. Drop readings that arrive
|
||||
# inside the sensor's effective polling interval so the Pi
|
||||
# agent can happily poll every 30 s while the log only
|
||||
# retains a row every 15–30 min. Inactive sensors also
|
||||
# dropped so disabled tanks don't clutter the log.
|
||||
if not sensor.active:
|
||||
skipped_interval += 1
|
||||
continue
|
||||
interval = sensor._fp_effective_poll_interval_minutes()
|
||||
if interval > 0 and sensor.last_reading_at:
|
||||
read_at = _parse_read_at(r.get('read_at'))
|
||||
elapsed = read_at - sensor.last_reading_at
|
||||
if elapsed < timedelta(minutes=interval):
|
||||
skipped_interval += 1
|
||||
continue
|
||||
|
||||
Reading.create({
|
||||
'sensor_id': sensor.id,
|
||||
'value': value,
|
||||
@@ -130,10 +148,11 @@ class FpIotIngestController(http.Controller):
|
||||
})
|
||||
accepted += 1
|
||||
|
||||
status = 200 if accepted else (404 if unknown_serials else 400)
|
||||
status = 200 if (accepted or skipped_interval) else (404 if unknown_serials else 400)
|
||||
payload = {
|
||||
'ok': accepted > 0,
|
||||
'ok': (accepted + skipped_interval) > 0,
|
||||
'accepted': accepted,
|
||||
'skipped': skipped_interval,
|
||||
}
|
||||
if unknown_serials:
|
||||
payload['unknown_serials'] = unknown_serials
|
||||
|
||||
@@ -23,6 +23,7 @@ _logger = logging.getLogger(__name__)
|
||||
def post_init_hook(env):
|
||||
_backfill_uuids(env)
|
||||
_backfill_sensor_types(env)
|
||||
_seed_entech_tanks_and_sensors(env)
|
||||
|
||||
|
||||
def _backfill_uuids(env):
|
||||
@@ -74,3 +75,98 @@ def _backfill_sensor_types(env):
|
||||
updated += 1
|
||||
_logger.info('fp.tank.sensor: set default sensor_type_id on %d records',
|
||||
updated)
|
||||
|
||||
|
||||
def _seed_entech_tanks_and_sensors(env):
|
||||
"""Sub 7 — seed 25 tanks (5 small + 20 big; 10 big inactive) with
|
||||
one temperature and one pH sensor each.
|
||||
|
||||
Idempotent: tanks keyed by `code` so re-runs skip existing rows.
|
||||
Sensors keyed by (tank_id, parameter_type) so duplicates aren't
|
||||
created if the admin added a temp/pH sensor manually after seed.
|
||||
|
||||
Opt-in via system parameter `fusion_plating_iot.seed_entech_tanks`.
|
||||
Default False so a fresh install doesn't auto-seed on anyone else.
|
||||
Admin sets it to '1' via Settings → Technical → System Parameters
|
||||
and re-runs the module upgrade (-u fusion_plating_iot) to trigger.
|
||||
"""
|
||||
Param = env['ir.config_parameter'].sudo()
|
||||
if Param.get_param('fusion_plating_iot.seed_entech_tanks', '0') != '1':
|
||||
return
|
||||
|
||||
# with_context(active_test=False) so the idempotency search sees
|
||||
# the inactive big tanks and doesn't try to recreate them on re-run.
|
||||
Tank = env['fusion.plating.tank'].with_context(active_test=False)
|
||||
Sensor = env['fp.tank.sensor'].with_context(active_test=False)
|
||||
Parameter = env['fusion.plating.bath.parameter']
|
||||
Facility = env['fusion.plating.facility']
|
||||
|
||||
facility = Facility.search([], limit=1)
|
||||
if not facility:
|
||||
_logger.warning('Sub 7 seed: no fusion.plating.facility found — '
|
||||
'skipping tank seed. Create a facility first.')
|
||||
return
|
||||
|
||||
temp_param = (
|
||||
Parameter.search([('code', '=', 'TEMP')], limit=1)
|
||||
or Parameter.search([('parameter_type', '=', 'temperature')], limit=1)
|
||||
)
|
||||
ph_param = (
|
||||
Parameter.search([('code', '=', 'PH')], limit=1)
|
||||
or Parameter.search([('parameter_type', '=', 'ph')], limit=1)
|
||||
)
|
||||
if not temp_param or not ph_param:
|
||||
_logger.warning('Sub 7 seed: temperature / pH bath parameters '
|
||||
'not found — skipping. Seed bath parameters first.')
|
||||
return
|
||||
|
||||
plan = []
|
||||
for i in range(1, 6):
|
||||
plan.append({'name': 'Small Tank #%d' % i, 'code': 'SMALL-%02d' % i,
|
||||
'active': True, 'sequence': 10 + i})
|
||||
for i in range(1, 21):
|
||||
plan.append({'name': 'Big Tank #%d' % i, 'code': 'BIG-%02d' % i,
|
||||
'active': (i <= 10), 'sequence': 20 + i})
|
||||
|
||||
tanks_created = 0
|
||||
sensors_created = 0
|
||||
for row in plan:
|
||||
tank = Tank.search([('code', '=', row['code'])], limit=1)
|
||||
if not tank:
|
||||
tank = Tank.create({
|
||||
'name': row['name'],
|
||||
'code': row['code'],
|
||||
'facility_id': facility.id,
|
||||
'sequence': row['sequence'],
|
||||
'active': row['active'],
|
||||
})
|
||||
tanks_created += 1
|
||||
# Temperature sensor
|
||||
if not Sensor.search_count([
|
||||
('tank_id', '=', tank.id),
|
||||
('parameter_id.parameter_type', '=', 'temperature'),
|
||||
]):
|
||||
Sensor.create({
|
||||
'name': '%s — Temperature' % tank.name,
|
||||
'tank_id': tank.id,
|
||||
'parameter_id': temp_param.id,
|
||||
'device_kind': 'ds18b20',
|
||||
'active': row['active'],
|
||||
})
|
||||
sensors_created += 1
|
||||
# pH sensor
|
||||
if not Sensor.search_count([
|
||||
('tank_id', '=', tank.id),
|
||||
('parameter_id.parameter_type', '=', 'ph'),
|
||||
]):
|
||||
Sensor.create({
|
||||
'name': '%s — pH' % tank.name,
|
||||
'tank_id': tank.id,
|
||||
'parameter_id': ph_param.id,
|
||||
'device_kind': 'ph',
|
||||
'active': row['active'],
|
||||
})
|
||||
sensors_created += 1
|
||||
|
||||
_logger.info('Sub 7 seed: created %d tanks + %d sensors',
|
||||
tanks_created, sensors_created)
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
from . import fp_sensor_type
|
||||
from . import fp_sensor_dashboard
|
||||
from . import res_company
|
||||
from . import res_config_settings
|
||||
from . import fp_tank_sensor
|
||||
from . import fp_tank_reading
|
||||
from . import fusion_plating_tank
|
||||
|
||||
@@ -163,6 +163,53 @@ class FpTankSensor(models.Model):
|
||||
'specific sensor.',
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Sub 7 — per-sensor polling interval
|
||||
#
|
||||
# Blank on the sensor = inherit res.company default. The effective
|
||||
# value gates the ingest endpoint so too-frequent readings are
|
||||
# dropped even when a Pi agent polls at a shorter cadence.
|
||||
# ------------------------------------------------------------------
|
||||
poll_interval_minutes = fields.Integer(
|
||||
string='Polling Interval (minutes)',
|
||||
help='How often a reading from this sensor should be stored. '
|
||||
'Leave blank to inherit the company-wide default. Readings '
|
||||
'that arrive inside this interval are dropped by the '
|
||||
'ingest endpoint — the database stays clean even if the '
|
||||
'Pi agent polls more often.',
|
||||
)
|
||||
poll_interval_display = fields.Char(
|
||||
string='Effective Interval',
|
||||
compute='_compute_poll_interval_display',
|
||||
help='Human-readable form of the sensor\'s effective polling '
|
||||
'interval. Shows "(override)" when the sensor carries its '
|
||||
'own value, "(default)" when it inherits.',
|
||||
)
|
||||
|
||||
@api.depends('poll_interval_minutes')
|
||||
def _compute_poll_interval_display(self):
|
||||
default = (self.env.company.x_fc_default_poll_interval_minutes
|
||||
or 30)
|
||||
for rec in self:
|
||||
if rec.poll_interval_minutes and rec.poll_interval_minutes > 0:
|
||||
rec.poll_interval_display = (
|
||||
'%d min (override)' % rec.poll_interval_minutes
|
||||
)
|
||||
else:
|
||||
rec.poll_interval_display = '%d min (default)' % default
|
||||
|
||||
def _fp_effective_poll_interval_minutes(self):
|
||||
"""Return the effective polling interval for this sensor.
|
||||
|
||||
Single lookup point. Call this rather than reading the raw
|
||||
field so later additions (per-tank override in Sub 8, per-
|
||||
customer override in Sub 6) only touch this helper.
|
||||
"""
|
||||
self.ensure_one()
|
||||
if self.poll_interval_minutes and self.poll_interval_minutes > 0:
|
||||
return self.poll_interval_minutes
|
||||
return self.env.company.x_fc_default_poll_interval_minutes or 30
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Effective target — resolves override → parameter default → 0
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
21
fusion_iot/fusion_plating_iot/models/res_company.py
Normal file
21
fusion_iot/fusion_plating_iot/models/res_company.py
Normal file
@@ -0,0 +1,21 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# Copyright 2026 Nexa Systems Inc.
|
||||
# License OPL-1 (Odoo Proprietary License v1.0)
|
||||
# Part of the Fusion Plating product family.
|
||||
#
|
||||
# Sub 7 — company-wide default for the IoT sensor polling interval.
|
||||
|
||||
from odoo import fields, models
|
||||
|
||||
|
||||
class ResCompany(models.Model):
|
||||
_inherit = 'res.company'
|
||||
|
||||
x_fc_default_poll_interval_minutes = fields.Integer(
|
||||
string='IoT default polling interval (minutes)',
|
||||
default=30,
|
||||
help='Applied to any fp.tank.sensor that does not set its own '
|
||||
'Polling Interval. Used by the ingest endpoint to drop '
|
||||
'readings that arrive inside the interval, keeping the '
|
||||
'database clean even when a Pi agent polls more often.',
|
||||
)
|
||||
19
fusion_iot/fusion_plating_iot/models/res_config_settings.py
Normal file
19
fusion_iot/fusion_plating_iot/models/res_config_settings.py
Normal file
@@ -0,0 +1,19 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# Copyright 2026 Nexa Systems Inc.
|
||||
# License OPL-1 (Odoo Proprietary License v1.0)
|
||||
# Part of the Fusion Plating product family.
|
||||
#
|
||||
# Sub 7 — Expose the IoT polling default on the Fusion Plating
|
||||
# Settings page so admins manage it alongside other plating settings.
|
||||
|
||||
from odoo import fields, models
|
||||
|
||||
|
||||
class ResConfigSettings(models.TransientModel):
|
||||
_inherit = 'res.config.settings'
|
||||
|
||||
x_fc_default_poll_interval_minutes = fields.Integer(
|
||||
related='company_id.x_fc_default_poll_interval_minutes',
|
||||
readonly=False,
|
||||
string='IoT default polling interval (minutes)',
|
||||
)
|
||||
@@ -30,6 +30,8 @@
|
||||
<field name="last_reading_at"/>
|
||||
<field name="last_reading_in_spec" widget="boolean_toggle"/>
|
||||
<field name="reading_count"/>
|
||||
<field name="poll_interval_minutes" optional="show"/>
|
||||
<field name="poll_interval_display" optional="show"/>
|
||||
<field name="active" column_invisible="1"/>
|
||||
</list>
|
||||
</field>
|
||||
@@ -110,6 +112,22 @@
|
||||
<field name="last_reading_in_spec" readonly="1" widget="boolean_toggle"/>
|
||||
</group>
|
||||
</group>
|
||||
<group string="IoT Polling (Sub 7)">
|
||||
<group>
|
||||
<field name="poll_interval_minutes"/>
|
||||
<field name="poll_interval_display" readonly="1"/>
|
||||
</group>
|
||||
<group>
|
||||
<div class="text-muted">
|
||||
<i class="fa fa-info-circle me-1"/>
|
||||
Leave the interval blank to inherit the
|
||||
company-wide default (configured under
|
||||
Settings → Fusion Plating → IoT). Readings
|
||||
arriving faster than the effective interval
|
||||
are dropped by the ingest endpoint.
|
||||
</div>
|
||||
</group>
|
||||
</group>
|
||||
</sheet>
|
||||
</form>
|
||||
</field>
|
||||
|
||||
@@ -0,0 +1,39 @@
|
||||
<?xml version="1.0" encoding="utf-8"?>
|
||||
<!--
|
||||
Copyright 2026 Nexa Systems Inc.
|
||||
License OPL-1 (Odoo Proprietary License v1.0)
|
||||
Part of the Fusion Plating product family.
|
||||
Sub 7 — IoT default polling interval setting.
|
||||
-->
|
||||
<odoo>
|
||||
|
||||
<record id="res_config_settings_view_form_fp_iot" model="ir.ui.view">
|
||||
<field name="name">res.config.settings.view.form.fp.iot</field>
|
||||
<field name="model">res.config.settings</field>
|
||||
<field name="inherit_id"
|
||||
ref="fusion_plating.res_config_settings_view_form_fp_core"/>
|
||||
<field name="arch" type="xml">
|
||||
<xpath expr="//app[@name='fusion_plating']" position="inside">
|
||||
<block title="IoT"
|
||||
name="fp_iot_settings"
|
||||
help="Defaults applied to physical sensors wired into
|
||||
Fusion Plating tanks. Per-sensor overrides live
|
||||
on each fp.tank.sensor record.">
|
||||
<setting id="fp_iot_default_poll_interval"
|
||||
string="Default Polling Interval"
|
||||
help="How often (in minutes) readings from a
|
||||
tank sensor are stored when the sensor
|
||||
itself does not specify its own interval.
|
||||
The ingest endpoint drops readings that
|
||||
arrive faster than the effective interval,
|
||||
so a Pi agent can freely poll more often
|
||||
without bloating the database.">
|
||||
<field name="x_fc_default_poll_interval_minutes"/>
|
||||
<span class="text-muted ms-2">minutes</span>
|
||||
</setting>
|
||||
</block>
|
||||
</xpath>
|
||||
</field>
|
||||
</record>
|
||||
|
||||
</odoo>
|
||||
@@ -372,7 +372,7 @@ rewrite code as new requirements surface. Each sub-project has its own design do
|
||||
| 4 | Contract Review (optional, per-part, settings-driven QA roster, QA-005 1:1 PDF) | **Shipped 2026-04-22** | 2i |
|
||||
| 5 | Order-line fields (fp.serial registry, auto job#, coating-scoped thickness dropdown, revision picker) | **Shipped 2026-04-22** | 5, 6, Q2 |
|
||||
| 6 | Contact Profiles & Communication Routing (sub-contacts + per-location notification lists + global contacts) | Pending | client transcript A/B/C |
|
||||
| 7 | IoT tuning (configurable polling interval 15–30 min, seed 6–10 tank sensors) | Pending | client transcript D |
|
||||
| 7 | IoT tuning (per-sensor polling interval + ingest rate-limit; entech seeded with 25 tanks / 50 sensors) | **Shipped 2026-04-22** | client transcript D |
|
||||
| 8 | Receiving / Inspection / QC flow restructure (split receiving vs inspection; racking crew inspects, not receiver) | Pending | client transcript E |
|
||||
| ∞ | First-off / last-off QC | Deferred | client transcript F |
|
||||
| ∞ | VEC machine auto-ingest (Word-format thickness report from network-connected XRF; different machine from Fischerscope) | Deferred | client transcript G |
|
||||
|
||||
@@ -0,0 +1,91 @@
|
||||
"""Sub 7 smoke test — runs inside odoo-shell on entech."""
|
||||
env = env
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
Param = env['ir.config_parameter'].sudo()
|
||||
Sensor = env['fp.tank.sensor']
|
||||
Tank = env['fusion.plating.tank']
|
||||
Reading = env['fp.tank.reading']
|
||||
|
||||
# ---- Field presence + helper -----------------------------------------
|
||||
assert 'poll_interval_minutes' in Sensor._fields
|
||||
assert 'poll_interval_display' in Sensor._fields
|
||||
assert 'x_fc_default_poll_interval_minutes' in env['res.company']._fields
|
||||
print('[OK] Fields present')
|
||||
|
||||
default = env.company.x_fc_default_poll_interval_minutes
|
||||
assert default == 30, f'default should be 30, got {default}'
|
||||
print(f'[OK] Company default polling = {default} min')
|
||||
|
||||
# Pick any existing sensor
|
||||
any_sensor = Sensor.search([], limit=1)
|
||||
assert any_sensor
|
||||
assert any_sensor._fp_effective_poll_interval_minutes() == 30
|
||||
print(f'[OK] Helper returns default for blank sensor')
|
||||
|
||||
any_sensor.poll_interval_minutes = 5
|
||||
any_sensor.invalidate_recordset()
|
||||
assert any_sensor._fp_effective_poll_interval_minutes() == 5
|
||||
assert '(override)' in any_sensor.poll_interval_display
|
||||
print(f'[OK] Override respected: {any_sensor.poll_interval_display}')
|
||||
|
||||
any_sensor.poll_interval_minutes = 0
|
||||
any_sensor.invalidate_recordset()
|
||||
assert '(default)' in any_sensor.poll_interval_display
|
||||
print(f'[OK] Blank interval shows default: {any_sensor.poll_interval_display}')
|
||||
|
||||
# ---- Seed the 25 tanks + 50 sensors via the hook ----------------------
|
||||
TankAll = Tank.with_context(active_test=False)
|
||||
SensorAll = Sensor.with_context(active_test=False)
|
||||
tanks_before = TankAll.search_count([])
|
||||
sensors_before = SensorAll.search_count([])
|
||||
Param.set_param('fusion_plating_iot.seed_entech_tanks', '1')
|
||||
from odoo.addons.fusion_plating_iot.hooks import _seed_entech_tanks_and_sensors
|
||||
_seed_entech_tanks_and_sensors(env)
|
||||
tanks_after = TankAll.search_count([])
|
||||
sensors_after = SensorAll.search_count([])
|
||||
print(f'[OK] Seed ran. Tanks: {tanks_before} → {tanks_after} (+{tanks_after - tanks_before})')
|
||||
print(f'[OK] Seed ran. Sensors: {sensors_before} → {sensors_after} (+{sensors_after - sensors_before})')
|
||||
assert tanks_after - tanks_before == 25, 'expected +25 tanks'
|
||||
assert sensors_after - sensors_before == 50, 'expected +50 sensors'
|
||||
|
||||
# Active / inactive breakdown
|
||||
small_active = TankAll.search_count([('code', 'like', 'SMALL-%'), ('active', '=', True)])
|
||||
big_active = TankAll.search_count([('code', 'like', 'BIG-%'), ('active', '=', True)])
|
||||
big_inactive = TankAll.search_count([('code', 'like', 'BIG-%'), ('active', '=', False)])
|
||||
assert small_active == 5, f'expected 5 small active, got {small_active}'
|
||||
assert big_active == 10, f'expected 10 big active, got {big_active}'
|
||||
assert big_inactive == 10, f'expected 10 big inactive, got {big_inactive}'
|
||||
print(f'[OK] Active/Inactive split: small={small_active}, big_active={big_active}, big_inactive={big_inactive}')
|
||||
|
||||
# Per-tank sensor counts
|
||||
sample = TankAll.search([('code', '=', 'BIG-01')], limit=1)
|
||||
sample_sensors = SensorAll.search([('tank_id', '=', sample.id)])
|
||||
types = sample_sensors.mapped('parameter_id.parameter_type')
|
||||
assert 'temperature' in types and 'ph' in types
|
||||
print(f'[OK] Big Tank #1 has sensors for: {types}')
|
||||
|
||||
# Idempotency — re-run seed, counts don't change
|
||||
_seed_entech_tanks_and_sensors(env)
|
||||
assert TankAll.search_count([]) == tanks_after, 'seed not idempotent (tanks)'
|
||||
assert SensorAll.search_count([]) == sensors_after, 'seed not idempotent (sensors)'
|
||||
print('[OK] Seed is idempotent')
|
||||
|
||||
# ---- Rate-limit: simulate two readings too close together -----------
|
||||
rate_sensor = Sensor.search([('code', '!=', False)], limit=1) if False else any_sensor
|
||||
rate_sensor.poll_interval_minutes = 30
|
||||
rate_sensor.last_reading_at = datetime.now() - timedelta(minutes=5) # 5 min ago
|
||||
print(f'[OK] Setup: interval=30min, last_reading=5min ago')
|
||||
# The ingest controller does the rate-limit. Simulate its check here.
|
||||
effective = rate_sensor._fp_effective_poll_interval_minutes()
|
||||
elapsed = datetime.now() - rate_sensor.last_reading_at
|
||||
assert elapsed < timedelta(minutes=effective), 'rate-limit check should fire'
|
||||
print('[OK] Rate-limit condition holds — reading would be skipped')
|
||||
|
||||
rate_sensor.last_reading_at = datetime.now() - timedelta(minutes=35) # 35 min ago
|
||||
elapsed = datetime.now() - rate_sensor.last_reading_at
|
||||
assert elapsed >= timedelta(minutes=effective), 'rate-limit should allow after interval'
|
||||
print('[OK] Post-interval condition holds — reading would be stored')
|
||||
|
||||
env.cr.rollback()
|
||||
print('\n=== SUB 7 SMOKE PASS — all assertions held ===')
|
||||
Reference in New Issue
Block a user