This commit is contained in:
gsinghpal
2026-05-16 13:18:52 -04:00
parent 191a9c82be
commit 9ebf89bde2
1080 changed files with 0 additions and 1197 deletions

View File

@@ -0,0 +1,31 @@
from . import test_depreciation_methods
from . import test_prorate
from . import test_salvage_value
from . import test_asset_anomaly_detection
from . import test_useful_life_predictor
from . import test_fusion_asset
from . import test_fusion_asset_depreciation_line
from . import test_fusion_asset_category
from . import test_fusion_asset_disposal
from . import test_fusion_asset_anomaly
from . import test_account_move_inherit
from . import test_fusion_asset_engine
from . import test_engine_integration
from . import test_assets_controller
from . import test_assets_adapter
from . import test_asset_tools
from . import test_assets_cron
from . import test_engine_property
from . import test_method_integration
from . import test_asset_book_values_mv
from . import test_performance_benchmarks
from . import test_create_asset_wizard
from . import test_disposal_wizard
from . import test_partial_sale_wizard
from . import test_depreciation_run_wizard
from . import test_migration_round_trip
from . import test_audit_report
from . import test_coexistence
from . import test_assets_tours
from . import test_perf_controller
from . import test_local_llm_compat

View File

@@ -0,0 +1,47 @@
from datetime import date
from odoo.tests.common import TransactionCase
from odoo.tests import tagged
@tagged('post_install', '-at_install')
class TestAccountMoveLineFusionAsset(TransactionCase):
def setUp(self):
super().setUp()
self.asset = self.env['fusion.asset'].create({
'name': 'Asset From Invoice',
'cost': 8000,
'acquisition_date': date(2026, 1, 1),
})
self.partner = self.env['res.partner'].create({'name': 'Vendor X'})
product = self.env['product.product'].create({'name': 'Test Asset Item'})
bill = self.env['account.move'].create({
'move_type': 'in_invoice',
'partner_id': self.partner.id,
'invoice_date': date(2026, 1, 1),
'invoice_line_ids': [(0, 0, {
'product_id': product.id,
'name': 'Test asset purchase',
'quantity': 1,
'price_unit': 8000,
})],
})
self.invoice_line = bill.invoice_line_ids[0]
def test_line_starts_without_asset_link(self):
self.assertFalse(self.invoice_line.fusion_asset_id)
self.assertEqual(self.invoice_line.fusion_asset_count, 0)
def test_link_invoice_line_to_asset(self):
self.invoice_line.fusion_asset_id = self.asset
self.assertEqual(self.invoice_line.fusion_asset_id, self.asset)
self.invoice_line.invalidate_recordset(['fusion_asset_count'])
self.assertEqual(self.invoice_line.fusion_asset_count, 1)
def test_action_open_fusion_asset_returns_window_action(self):
self.invoice_line.fusion_asset_id = self.asset
action = self.invoice_line.action_open_fusion_asset()
self.assertEqual(action['res_model'], 'fusion.asset')
self.assertEqual(action['res_id'], self.asset.id)
self.assertEqual(action['view_mode'], 'form')

View File

@@ -0,0 +1,71 @@
from odoo.tests.common import TransactionCase
from odoo.tests import tagged
from odoo.addons.fusion_accounting_assets.services.anomaly_detection import (
detect_schedule_variance, detect_low_utilization, AssetAnomaly,
)
@tagged('post_install', '-at_install')
class TestAssetAnomalyDetection(TransactionCase):
def test_schedule_variance_within_threshold_returns_none(self):
# 5% variance < 10% threshold
result = detect_schedule_variance(
asset_id=1, asset_name='Truck', expected_accumulated=10000,
actual_accumulated=10500,
)
self.assertIsNone(result)
def test_schedule_variance_behind_schedule_low_severity(self):
# 15% behind: low severity, behind_schedule
result = detect_schedule_variance(
asset_id=1, asset_name='Truck', expected_accumulated=10000,
actual_accumulated=8500,
)
self.assertIsNotNone(result)
self.assertEqual(result.anomaly_type, 'behind_schedule')
self.assertEqual(result.severity, 'low')
def test_schedule_variance_ahead_high_severity(self):
# 60% ahead: high severity
result = detect_schedule_variance(
asset_id=2, asset_name='Server', expected_accumulated=10000,
actual_accumulated=16000,
)
self.assertIsNotNone(result)
self.assertEqual(result.anomaly_type, 'ahead_of_schedule')
self.assertEqual(result.severity, 'high')
def test_schedule_variance_zero_expected_returns_none(self):
result = detect_schedule_variance(
asset_id=1, asset_name='Truck', expected_accumulated=0,
actual_accumulated=500,
)
self.assertIsNone(result)
def test_low_utilization_flags_when_underused(self):
# 60% deficit -> high severity
result = detect_low_utilization(
asset_id=3, asset_name='Mill', expected_units=1000, actual_units=400,
)
self.assertIsNotNone(result)
self.assertEqual(result.anomaly_type, 'low_utilization')
self.assertEqual(result.severity, 'high')
def test_low_utilization_within_tolerance_returns_none(self):
# 95% used: within 10% tolerance
result = detect_low_utilization(
asset_id=3, asset_name='Mill', expected_units=1000, actual_units=950,
)
self.assertIsNone(result)
def test_anomaly_to_dict_round_trip(self):
anomaly = AssetAnomaly(
asset_id=1, asset_name='X', anomaly_type='behind_schedule',
severity='medium', expected=100.0, actual=70.0, variance_pct=30.0,
detail='example',
)
d = anomaly.to_dict()
self.assertEqual(d['asset_id'], 1)
self.assertEqual(d['anomaly_type'], 'behind_schedule')
self.assertEqual(d['severity'], 'medium')

View File

@@ -0,0 +1,29 @@
"""Tests for the per-asset book value MV."""
from datetime import date
from odoo.tests.common import TransactionCase
from odoo.tests import tagged
@tagged('post_install', '-at_install')
class TestAssetBookValuesMV(TransactionCase):
def test_mv_exists_and_is_queryable(self):
self.env['fusion.asset.book.values.mv']._refresh(concurrently=False)
rows = self.env['fusion.asset.book.values.mv'].search([], limit=10)
self.assertIsNotNone(rows)
def test_mv_includes_new_asset_after_refresh(self):
asset = self.env['fusion.asset'].create({
'name': 'MV Test', 'cost': 5000, 'salvage_value': 500,
'acquisition_date': date(2026, 1, 1),
'method': 'straight_line', 'useful_life_years': 5,
})
self.env.flush_all()
self.env['fusion.asset.book.values.mv']._refresh(concurrently=False)
mv_row = self.env['fusion.asset.book.values.mv'].search([
('asset_id', '=', asset.id),
], limit=1)
self.assertTrue(mv_row)
self.assertAlmostEqual(mv_row.book_value, 5000, places=2)

View File

@@ -0,0 +1,56 @@
"""Tests for the 5 fusion-asset AI tools."""
from datetime import date
from odoo.tests import tagged
from odoo.tests.common import TransactionCase
from odoo.addons.fusion_accounting_ai.services.tools import asset_management as tools
@tagged('post_install', '-at_install')
class TestFusionAssetTools(TransactionCase):
def test_fusion_list_assets(self):
self.env['fusion.asset'].create({
'name': 'Tool Test', 'cost': 1000,
'acquisition_date': date(2026, 1, 1),
'method': 'straight_line', 'useful_life_years': 4,
})
result = tools.fusion_list_assets(self.env, {'company_id': self.env.company.id})
self.assertGreaterEqual(result.get('count', 0), 1)
def test_fusion_get_asset_detail(self):
asset = self.env['fusion.asset'].create({
'name': 'Detail Test', 'cost': 1500,
'acquisition_date': date(2026, 1, 1),
'method': 'straight_line', 'useful_life_years': 4,
})
result = tools.fusion_get_asset_detail(self.env, {'asset_id': asset.id})
self.assertEqual(result['asset']['name'], 'Detail Test')
def test_fusion_compute_schedule(self):
asset = self.env['fusion.asset'].create({
'name': 'Schedule Test', 'cost': 2000,
'acquisition_date': date(2026, 1, 1),
'method': 'straight_line', 'useful_life_years': 4,
})
result = tools.fusion_compute_asset_schedule(self.env, {'asset_id': asset.id})
self.assertEqual(result['lines_created'], 4)
def test_fusion_suggest_useful_life(self):
self.env['ir.config_parameter'].sudo().search([
('key', 'in', ['fusion_accounting.provider.asset_useful_life',
'fusion_accounting.provider.default'])
]).unlink()
result = tools.fusion_suggest_asset_useful_life(self.env, {
'description': 'desk',
})
self.assertEqual(result['useful_life_years'], 7)
def test_tools_registered_in_dispatch(self):
from odoo.addons.fusion_accounting_ai.services.tools import TOOL_DISPATCH
for tool_name in ['fusion_list_assets', 'fusion_get_asset_detail',
'fusion_compute_asset_schedule', 'fusion_dispose_asset',
'fusion_suggest_asset_useful_life']:
self.assertIn(tool_name, TOOL_DISPATCH)

View File

@@ -0,0 +1,40 @@
"""AssetsAdapter wiring tests — fusion-mode dispatch."""
from datetime import date
from odoo.tests import tagged
from odoo.tests.common import TransactionCase
from odoo.addons.fusion_accounting_ai.services.data_adapters.assets import (
AssetsAdapter,
)
@tagged('post_install', '-at_install')
class TestAssetsAdapter(TransactionCase):
def setUp(self):
super().setUp()
self.adapter = AssetsAdapter(self.env)
def test_list_assets_via_fusion(self):
self.env['fusion.asset'].create({
'name': 'Adapter Test', 'cost': 1000,
'acquisition_date': date(2026, 1, 1),
'method': 'straight_line', 'useful_life_years': 4,
})
result = self.adapter.list_assets_via_fusion(company_id=self.env.company.id)
self.assertGreaterEqual(result['count'], 1)
def test_suggest_useful_life_via_fusion_uses_templated_fallback(self):
self.env['ir.config_parameter'].sudo().search([
('key', 'in', ['fusion_accounting.provider.asset_useful_life',
'fusion_accounting.provider.default'])
]).unlink()
result = self.adapter.suggest_useful_life_via_fusion(description='laptop')
self.assertEqual(result['useful_life_years'], 4)
self.assertEqual(result['depreciation_method'], 'straight_line')
def test_dispose_asset_via_community_returns_error(self):
result = self.adapter.dispose_asset_via_community(asset_id=1, sale_amount=100)
self.assertIn('error', result)

View File

@@ -0,0 +1,103 @@
"""Controller tests using HttpCase."""
import json
from datetime import date
from odoo.tests import tagged
from odoo.tests.common import HttpCase, new_test_user
@tagged('post_install', '-at_install')
class TestAssetsController(HttpCase):
def setUp(self):
super().setUp()
self.user = new_test_user(
self.env, login='assets_test_user',
groups='base.group_user,account.group_account_invoice',
)
def _jsonrpc(self, endpoint, params):
self.authenticate('assets_test_user', 'assets_test_user')
url = f'/fusion/assets/{endpoint}'
body = {'jsonrpc': '2.0', 'method': 'call', 'params': params, 'id': 1}
response = self.url_open(
url, data=json.dumps(body),
headers={'Content-Type': 'application/json'},
)
self.assertEqual(
response.status_code, 200,
f"{endpoint} returned {response.status_code}: {response.text[:300]}",
)
result = response.json()
if 'error' in result:
self.fail(f"{endpoint} errored: {result['error']}")
return result.get('result', {})
def test_list_returns_dict(self):
result = self._jsonrpc('list', {'company_id': self.env.company.id})
self.assertIn('assets', result)
self.assertIn('total', result)
def test_get_detail_returns_asset(self):
asset = self.env['fusion.asset'].create({
'name': 'Ctrl Test Asset', 'cost': 5000,
'acquisition_date': date(2026, 1, 1),
'method': 'straight_line', 'useful_life_years': 5,
})
result = self._jsonrpc('get_detail', {'asset_id': asset.id})
self.assertEqual(result['asset']['id'], asset.id)
self.assertIn('depreciation_lines', result)
def test_compute_schedule_creates_lines(self):
asset = self.env['fusion.asset'].create({
'name': 'CompTest', 'cost': 4000,
'acquisition_date': date(2026, 1, 1),
'method': 'straight_line', 'useful_life_years': 4,
})
result = self._jsonrpc('compute_schedule', {'asset_id': asset.id})
self.assertEqual(result['lines_created'], 4)
def test_post_depreciation_after_running(self):
asset = self.env['fusion.asset'].create({
'name': 'PostTest', 'cost': 3000,
'acquisition_date': date(2026, 1, 1),
'in_service_date': date(2026, 1, 1),
'method': 'straight_line', 'useful_life_years': 3,
})
self.env['fusion.asset.engine'].compute_depreciation_schedule(asset)
asset.action_set_running()
result = self._jsonrpc('post_depreciation', {'asset_id': asset.id})
self.assertEqual(result['posted_count'], 1)
def test_dispose_marks_asset_disposed(self):
asset = self.env['fusion.asset'].create({
'name': 'DispTest', 'cost': 6000,
'acquisition_date': date(2026, 1, 1),
'in_service_date': date(2026, 1, 1),
'method': 'straight_line', 'useful_life_years': 3,
})
self.env['fusion.asset.engine'].compute_depreciation_schedule(asset)
asset.action_set_running()
result = self._jsonrpc('dispose', {
'asset_id': asset.id, 'sale_amount': 4000,
'sale_date': '2027-06-01', 'disposal_type': 'sale',
})
self.assertIn('disposal_id', result)
asset.invalidate_recordset(['state'])
self.assertEqual(asset.state, 'disposed')
def test_get_anomalies_returns_list(self):
result = self._jsonrpc('get_anomalies', {'company_id': self.env.company.id})
self.assertIn('anomalies', result)
def test_suggest_useful_life_returns_dict(self):
result = self._jsonrpc('suggest_useful_life', {'description': 'Dell laptop'})
self.assertIn('useful_life_years', result)
self.assertIn('depreciation_method', result)
self.assertEqual(result['useful_life_years'], 4)
def test_get_partner_history(self):
partner = self.env['res.partner'].create({'name': 'History Test Partner'})
result = self._jsonrpc('get_partner_history', {'partner_id': partner.id})
self.assertEqual(result['partner_id'], partner.id)

View File

@@ -0,0 +1,28 @@
"""Cron handler smoke tests."""
from datetime import date
from odoo.tests import tagged
from odoo.tests.common import TransactionCase
@tagged('post_install', '-at_install')
class TestFusionAssetsCron(TransactionCase):
def setUp(self):
super().setUp()
self.cron = self.env['fusion.assets.cron']
self.asset = self.env['fusion.asset'].create({
'name': 'Cron Test', 'cost': 4000,
'acquisition_date': date(2026, 1, 1),
'in_service_date': date(2026, 1, 1),
'method': 'straight_line', 'useful_life_years': 4,
})
self.env['fusion.asset.engine'].compute_depreciation_schedule(self.asset)
self.asset.action_set_running()
def test_cron_post_due_depreciation_runs(self):
self.cron._cron_post_due_depreciation()
def test_cron_anomaly_scan_runs(self):
self.cron._cron_anomaly_scan()

View File

@@ -0,0 +1,28 @@
"""Python wrappers that run the OWL tours via HttpCase.start_tour.
Tours require an HTTP server + headless browser. They are tagged with
'tour' so they can be excluded from fast unit-test runs and selected
explicitly when CI has the right infra (chromium + xvfb / websocket-client).
"""
from odoo.tests.common import HttpCase
from odoo.tests import tagged
@tagged('post_install', '-at_install', 'tour')
class TestAssetsTours(HttpCase):
def test_smoke_tour(self):
self.start_tour("/odoo", "fusion_assets_smoke", login="admin")
def test_list_tour(self):
self.start_tour("/odoo", "fusion_assets_list", login="admin")
def test_categories_tour(self):
self.start_tour("/odoo", "fusion_assets_categories", login="admin")
def test_anomalies_tour(self):
self.start_tour("/odoo", "fusion_assets_anomalies", login="admin")
def test_depreciation_wizard_tour(self):
self.start_tour("/odoo", "fusion_assets_depreciation_wizard", login="admin")

View File

@@ -0,0 +1,18 @@
from odoo.tests.common import TransactionCase
from odoo.tests import tagged
@tagged('post_install', '-at_install')
class TestAuditReport(TransactionCase):
def test_report_renders(self):
wizard = self.env['fusion.migration.wizard'].create({})
try:
pdf, content_type = self.env['ir.actions.report'].sudo()._render_qweb_pdf(
'fusion_accounting_assets.migration_audit_template',
res_ids=[wizard.id], data={},
)
# PDF or HTML both ok (wkhtmltopdf might be missing on dev VM)
self.assertGreater(len(pdf), 100)
except Exception as e:
self.skipTest(f"PDF render failed (likely wkhtmltopdf missing): {e}")

View File

@@ -0,0 +1,38 @@
"""Coexistence tests: fusion_accounting_assets menu only visible when
Enterprise account_asset is NOT installed."""
from odoo.tests.common import TransactionCase
from odoo.tests import tagged
@tagged('post_install', '-at_install')
class TestAssetsCoexistence(TransactionCase):
def setUp(self):
super().setUp()
self.coex_group = self.env.ref(
'fusion_accounting_core.group_fusion_show_when_enterprise_absent',
raise_if_not_found=False,
)
self.assertIsNotNone(self.coex_group, "Coexistence group must exist")
def test_engine_always_available(self):
"""Engine is registered regardless of Enterprise install state."""
self.assertIn('fusion.asset.engine', self.env.registry)
def test_menu_gated_by_coexistence_group(self):
menu = self.env.ref('fusion_accounting_assets.menu_fusion_assets_root',
raise_if_not_found=False)
if not menu:
self.skipTest("Menu not loaded")
menu_groups = getattr(menu, 'group_ids', None) or menu.groups_id
self.assertIn(self.coex_group, menu_groups,
"Asset root menu must require the coexistence group")
def test_categories_menu_gated(self):
menu = self.env.ref('fusion_accounting_assets.menu_fusion_asset_categories',
raise_if_not_found=False)
if not menu:
self.skipTest("Menu not loaded")
menu_groups = getattr(menu, 'group_ids', None) or menu.groups_id
self.assertIn(self.coex_group, menu_groups)

View File

@@ -0,0 +1,62 @@
from datetime import date
from odoo.tests import tagged
from odoo.tests.common import TransactionCase
@tagged('post_install', '-at_install')
class TestCreateAssetWizard(TransactionCase):
def setUp(self):
super().setUp()
self.env['ir.config_parameter'].sudo().search([
('key', 'in', ['fusion_accounting.provider.asset_useful_life',
'fusion_accounting.provider.default'])
]).unlink()
def test_create_minimal_asset(self):
wizard = self.env['fusion.create.asset.wizard'].create({
'name': 'Test Asset',
'cost': 5000,
'method': 'straight_line',
'useful_life_years': 5,
'acquisition_date': date(2026, 1, 1),
'source_invoice_line_id': False,
})
action = wizard.action_create_asset()
self.assertEqual(action['res_model'], 'fusion.asset')
asset = self.env['fusion.asset'].browse(action['res_id'])
self.assertEqual(asset.name, 'Test Asset')
self.assertEqual(asset.cost, 5000)
def test_ai_suggest_fills_fields(self):
wizard = self.env['fusion.create.asset.wizard'].create({
'name': 'Dell laptop',
'cost': 2000,
'method': 'straight_line',
'useful_life_years': 5,
'acquisition_date': date(2026, 1, 1),
})
wizard.action_ai_suggest()
self.assertEqual(wizard.ai_suggested_years, 4)
self.assertEqual(wizard.useful_life_years, 4)
def test_category_onchange_pre_fills(self):
category = self.env['fusion.asset.category'].create({
'name': 'Test Category',
'method': 'declining_balance',
'useful_life_years': 7,
'declining_rate_pct': 25.0,
'salvage_value_pct': 10.0,
})
wizard = self.env['fusion.create.asset.wizard'].new({
'name': 'Test', 'cost': 10000,
'method': 'straight_line', 'useful_life_years': 5,
'acquisition_date': date(2026, 1, 1),
'category_id': category.id,
})
wizard._onchange_category_id()
self.assertEqual(wizard.method, 'declining_balance')
self.assertEqual(wizard.useful_life_years, 7)
self.assertEqual(wizard.declining_rate_pct, 25.0)
self.assertAlmostEqual(wizard.salvage_value, 1000, places=2)

View File

@@ -0,0 +1,88 @@
from odoo.tests.common import TransactionCase
from odoo.tests import tagged
from odoo.addons.fusion_accounting_assets.services.depreciation_methods import (
straight_line, declining_balance, units_of_production,
)
@tagged('post_install', '-at_install')
class TestStraightLine(TransactionCase):
def test_total_equals_cost_minus_salvage(self):
steps = straight_line(cost=10000, salvage_value=1000, n_periods=5)
total = sum(s.period_amount for s in steps)
self.assertAlmostEqual(total, 9000, places=2)
def test_per_period_equal_except_last(self):
steps = straight_line(cost=10000, salvage_value=0, n_periods=4)
self.assertEqual([s.period_amount for s in steps], [2500.0] * 4)
def test_last_period_absorbs_rounding(self):
steps = straight_line(cost=10000, salvage_value=0, n_periods=3)
total = sum(s.period_amount for s in steps)
self.assertAlmostEqual(total, 10000, places=2)
def test_zero_periods_returns_empty(self):
self.assertEqual(straight_line(cost=10000, n_periods=0), [])
def test_book_value_decreasing(self):
steps = straight_line(cost=10000, salvage_value=1000, n_periods=5)
for i in range(1, len(steps)):
self.assertLess(steps[i].book_value_at_end, steps[i - 1].book_value_at_end)
@tagged('post_install', '-at_install')
class TestDecliningBalance(TransactionCase):
def test_total_does_not_exceed_depreciable(self):
steps = declining_balance(cost=10000, salvage_value=1000, n_periods=10, rate=0.20)
total = sum(s.period_amount for s in steps)
self.assertLessEqual(total, 9000.01)
def test_does_not_go_below_salvage(self):
steps = declining_balance(cost=10000, salvage_value=1000, n_periods=10, rate=0.50)
for s in steps:
self.assertGreaterEqual(s.book_value_at_end, 999.99)
def test_zero_rate_returns_empty(self):
self.assertEqual(declining_balance(cost=10000, n_periods=5, rate=0), [])
def test_pathological_100pct_rate_one_period(self):
steps = declining_balance(cost=10000, salvage_value=500, n_periods=10, rate=1.0)
self.assertEqual(len(steps), 1)
self.assertAlmostEqual(steps[0].period_amount, 9500, places=2)
@tagged('post_install', '-at_install')
class TestUnitsOfProduction(TransactionCase):
def test_total_proportional_to_units_used(self):
steps = units_of_production(
cost=20000, salvage_value=2000,
total_units_expected=10000,
units_per_period=[1000, 2000, 3000, 4000],
)
total = sum(s.period_amount for s in steps)
self.assertAlmostEqual(total, 18000, places=1)
def test_partial_use_partial_depreciation(self):
steps = units_of_production(
cost=10000, salvage_value=0,
total_units_expected=1000,
units_per_period=[200],
)
self.assertAlmostEqual(steps[0].period_amount, 2000, places=2)
def test_zero_total_units_returns_empty(self):
self.assertEqual(
units_of_production(cost=10000, total_units_expected=0, units_per_period=[100]),
[],
)
def test_does_not_overshoot_salvage(self):
steps = units_of_production(
cost=10000, salvage_value=1000,
total_units_expected=1000,
units_per_period=[2000],
)
self.assertAlmostEqual(steps[0].period_amount, 9000, places=2)

View File

@@ -0,0 +1,43 @@
from datetime import date
from odoo.tests import tagged
from odoo.tests.common import TransactionCase
@tagged('post_install', '-at_install')
class TestDepreciationRunWizard(TransactionCase):
def test_run_all_running_posts_due_periods(self):
for amt in [3000, 5000]:
asset = self.env['fusion.asset'].create({
'name': f'Run Test {amt}', 'cost': amt,
'acquisition_date': date(2026, 1, 1),
'in_service_date': date(2026, 1, 1),
'method': 'straight_line', 'useful_life_years': 3,
})
self.env['fusion.asset.engine'].compute_depreciation_schedule(asset)
asset.action_set_running()
wizard = self.env['fusion.depreciation.run.wizard'].create({
'period_date': date(2030, 12, 31),
'state_filter': 'all_running',
})
wizard.action_run()
self.assertEqual(wizard.state, 'done')
self.assertGreater(wizard.posted_count, 0)
def test_run_selected_posts_only_selected(self):
asset = self.env['fusion.asset'].create({
'name': 'Selected Test', 'cost': 1000,
'acquisition_date': date(2026, 1, 1),
'in_service_date': date(2026, 1, 1),
'method': 'straight_line', 'useful_life_years': 3,
})
self.env['fusion.asset.engine'].compute_depreciation_schedule(asset)
asset.action_set_running()
wizard = self.env['fusion.depreciation.run.wizard'].create({
'period_date': date(2030, 12, 31),
'state_filter': 'selected',
'asset_ids': [(6, 0, [asset.id])],
})
wizard.action_run()
self.assertEqual(wizard.state, 'done')

View File

@@ -0,0 +1,50 @@
from datetime import date
from odoo.tests import tagged
from odoo.tests.common import TransactionCase
@tagged('post_install', '-at_install')
class TestDisposalWizard(TransactionCase):
def setUp(self):
super().setUp()
self.asset = self.env['fusion.asset'].create({
'name': 'Disposal Test Asset',
'cost': 6000,
'acquisition_date': date(2026, 1, 1),
'in_service_date': date(2026, 1, 1),
'method': 'straight_line', 'useful_life_years': 3,
})
self.env['fusion.asset.engine'].compute_depreciation_schedule(self.asset)
self.asset.action_set_running()
def test_default_loads_active_asset(self):
wizard = self.env['fusion.disposal.wizard'].with_context(
active_model='fusion.asset', active_id=self.asset.id,
).create({})
self.assertEqual(wizard.asset_id, self.asset)
def test_action_dispose_marks_asset_disposed(self):
wizard = self.env['fusion.disposal.wizard'].create({
'asset_id': self.asset.id,
'disposal_type': 'sale',
'sale_amount': 4000,
'disposal_date': date(2026, 6, 1),
})
wizard.action_dispose()
self.asset.invalidate_recordset(['state'])
self.assertEqual(self.asset.state, 'disposed')
def test_compute_gain_loss_sale(self):
wizard = self.env['fusion.disposal.wizard'].create({
'asset_id': self.asset.id,
'disposal_type': 'sale',
'sale_amount': 7000,
})
wizard._compute_gain_loss()
self.assertAlmostEqual(
wizard.estimated_gain_loss,
7000 - self.asset.book_value,
places=2,
)

View File

@@ -0,0 +1,151 @@
"""End-to-end engine integration tests.
Each test creates a complete realistic asset (with category and accounts),
runs the engine through a full lifecycle, and asserts both the model state
and the journal entries (where category accounts are configured).
"""
from datetime import date
from odoo.tests.common import TransactionCase
from odoo.tests import tagged
from odoo.exceptions import ValidationError
@tagged('post_install', '-at_install', 'integration')
class TestAssetEngineIntegration(TransactionCase):
def setUp(self):
super().setUp()
self.engine = self.env['fusion.asset.engine']
Account = self.env['account.account']
company_id = self.env.company.id
self.expense_account = Account.search([
('account_type', '=', 'expense_depreciation'),
('company_ids', 'in', company_id),
], limit=1)
if not self.expense_account:
self.expense_account = Account.create({
'name': 'Test Depreciation Expense',
'code': '7180',
'account_type': 'expense_depreciation',
'company_ids': [(6, 0, [company_id])],
})
self.dep_account = Account.search([
('account_type', '=', 'asset_fixed'),
('company_ids', 'in', company_id),
], limit=1)
if not self.dep_account:
self.dep_account = Account.create({
'name': 'Test Accumulated Depreciation',
'code': '1690',
'account_type': 'asset_fixed',
'company_ids': [(6, 0, [company_id])],
})
self.category = self.env['fusion.asset.category'].create({
'name': 'Test Category',
'method': 'straight_line',
'useful_life_years': 5,
'asset_account_id': self.dep_account.id,
'depreciation_account_id': self.dep_account.id,
'expense_account_id': self.expense_account.id,
})
def _make_asset(self, **kwargs):
defaults = {
'name': 'Integration Asset',
'cost': 12000,
'salvage_value': 0,
'acquisition_date': date(2026, 1, 1),
'in_service_date': date(2026, 1, 1),
'method': 'straight_line',
'useful_life_years': 4,
'category_id': self.category.id,
}
defaults.update(kwargs)
return self.env['fusion.asset'].create(defaults)
def test_full_lifecycle_straight_line(self):
asset = self._make_asset()
self.engine.compute_depreciation_schedule(asset)
self.assertEqual(len(asset.depreciation_line_ids), 4)
self.assertAlmostEqual(
sum(asset.depreciation_line_ids.mapped('amount')), 12000, places=2,
)
asset.action_set_running()
for _i in range(2):
result = self.engine.post_depreciation_entry(asset)
self.assertEqual(result['posted_count'], 1)
asset.invalidate_recordset(['book_value', 'total_depreciated'])
self.assertAlmostEqual(asset.total_depreciated, 6000, places=2)
def test_post_creates_journal_entry_when_accounts_configured(self):
asset = self._make_asset()
self.engine.compute_depreciation_schedule(asset)
asset.action_set_running()
self.engine.post_depreciation_entry(asset)
first = asset.depreciation_line_ids.sorted('period_index')[0]
self.assertTrue(first.move_id, "Expected journal entry on posted line")
moves = first.move_id
self.assertAlmostEqual(
sum(moves.line_ids.mapped('debit')),
sum(moves.line_ids.mapped('credit')),
places=2,
)
def test_dispose_caps_future_lines(self):
asset = self._make_asset()
self.engine.compute_depreciation_schedule(asset)
asset.action_set_running()
self.engine.post_depreciation_entry(asset)
self.engine.dispose_asset(
asset, sale_amount=5000, sale_date=date(2027, 6, 1),
)
self.assertEqual(asset.state, 'disposed')
unposted = asset.depreciation_line_ids.filtered(lambda l: not l.is_posted)
for line in unposted:
self.assertLessEqual(line.scheduled_date, date(2027, 6, 1))
def test_dispose_records_correct_book_value(self):
asset = self._make_asset()
self.engine.compute_depreciation_schedule(asset)
asset.action_set_running()
for _i in range(2):
self.engine.post_depreciation_entry(asset)
result = self.engine.dispose_asset(
asset, sale_amount=8000, sale_date=date(2028, 6, 1),
)
# Book value at disposal = cost - accumulated = 12000 - 6000 = 6000.
self.assertAlmostEqual(result['book_value_at_disposal'], 6000, places=2)
# Gain = 8000 - 6000 = 2000.
self.assertAlmostEqual(result['gain_loss_amount'], 2000, places=2)
def test_partial_sale_30pct(self):
asset = self._make_asset(cost=10000, salvage_value=0)
self.engine.compute_depreciation_schedule(asset)
asset.action_set_running()
result = self.engine.partial_sale(
asset, sold_amount=3500, sold_qty=0.3,
sale_date=date(2027, 1, 1),
)
asset.invalidate_recordset(['cost'])
self.assertAlmostEqual(asset.cost, 7000, places=2)
child = self.env['fusion.asset'].browse(result['child_asset_id'])
self.assertAlmostEqual(child.cost, 3000, places=2)
self.assertEqual(child.state, 'disposed')
# Child has no posted depreciation; book_value at disposal = 3000.
# Gain = 3500 - 3000 = 500.
self.assertAlmostEqual(result['gain_loss_amount'], 500, places=0)
def test_pause_then_resume_lifecycle(self):
asset = self._make_asset()
self.engine.compute_depreciation_schedule(asset)
asset.action_set_running()
self.engine.post_depreciation_entry(asset)
self.engine.pause_asset(asset)
with self.assertRaises(ValidationError):
self.engine.post_depreciation_entry(asset)
self.engine.resume_asset(asset)
result = self.engine.post_depreciation_entry(asset)
self.assertEqual(result['posted_count'], 1)

View File

@@ -0,0 +1,101 @@
"""Property-based invariant tests for the asset engine.
Hypothesis generates random inputs; we assert mathematical invariants
that must hold regardless of input."""
from hypothesis import given, settings, strategies as st, HealthCheck
from odoo.tests.common import TransactionCase
from odoo.tests import tagged
from odoo.addons.fusion_accounting_assets.services.depreciation_methods import (
straight_line, declining_balance, units_of_production,
)
@tagged('post_install', '-at_install', 'property_based')
class TestDepreciationInvariants(TransactionCase):
@given(
cost=st.floats(min_value=100.0, max_value=1000000.0,
allow_nan=False, allow_infinity=False),
salvage_pct=st.floats(min_value=0.0, max_value=0.5,
allow_nan=False, allow_infinity=False),
n_periods=st.integers(min_value=1, max_value=40),
)
@settings(max_examples=80, deadline=2000,
suppress_health_check=[HealthCheck.function_scoped_fixture])
def test_straight_line_total_equals_cost_minus_salvage(self, cost, salvage_pct, n_periods):
cost = round(cost, 2)
salvage = round(cost * salvage_pct, 2)
steps = straight_line(cost=cost, salvage_value=salvage, n_periods=n_periods)
total = sum(s.period_amount for s in steps)
# Within 1c rounding tolerance
self.assertAlmostEqual(
total, cost - salvage, places=1,
msg=f"cost={cost}, salvage={salvage}, n={n_periods}, total={total:.2f}",
)
@given(
cost=st.floats(min_value=100.0, max_value=1000000.0,
allow_nan=False, allow_infinity=False),
salvage_pct=st.floats(min_value=0.0, max_value=0.5,
allow_nan=False, allow_infinity=False),
n_periods=st.integers(min_value=1, max_value=20),
)
@settings(max_examples=50, deadline=2000,
suppress_health_check=[HealthCheck.function_scoped_fixture])
def test_straight_line_book_value_decreasing(self, cost, salvage_pct, n_periods):
cost = round(cost, 2)
salvage = round(cost * salvage_pct, 2)
steps = straight_line(cost=cost, salvage_value=salvage, n_periods=n_periods)
for i in range(1, len(steps)):
self.assertLessEqual(
steps[i].book_value_at_end,
steps[i - 1].book_value_at_end + 0.01,
)
@given(
cost=st.floats(min_value=1000.0, max_value=100000.0,
allow_nan=False, allow_infinity=False),
salvage_pct=st.floats(min_value=0.0, max_value=0.3,
allow_nan=False, allow_infinity=False),
n_periods=st.integers(min_value=2, max_value=20),
rate=st.floats(min_value=0.05, max_value=0.5,
allow_nan=False, allow_infinity=False),
)
@settings(max_examples=50, deadline=3000,
suppress_health_check=[HealthCheck.function_scoped_fixture])
def test_declining_balance_never_below_salvage(self, cost, salvage_pct, n_periods, rate):
cost = round(cost, 2)
salvage = round(cost * salvage_pct, 2)
steps = declining_balance(
cost=cost, salvage_value=salvage,
n_periods=n_periods, rate=rate,
)
for s in steps:
self.assertGreaterEqual(
s.book_value_at_end, salvage - 0.01,
msg=f"cost={cost}, salvage={salvage}, rate={rate}, step={s}",
)
@given(
cost=st.floats(min_value=1000.0, max_value=100000.0,
allow_nan=False, allow_infinity=False),
total_units=st.floats(min_value=100.0, max_value=10000.0,
allow_nan=False, allow_infinity=False),
n_periods=st.integers(min_value=1, max_value=10),
)
@settings(max_examples=30, deadline=2000,
suppress_health_check=[HealthCheck.function_scoped_fixture])
def test_units_of_production_total_at_full_use_equals_depreciable(self, cost, total_units, n_periods):
cost = round(cost, 2)
salvage = 0.0
# Distribute total_units evenly across periods
per_period = total_units / n_periods
steps = units_of_production(
cost=cost, salvage_value=salvage,
total_units_expected=total_units,
units_per_period=[per_period] * n_periods,
)
total = sum(s.period_amount for s in steps)
self.assertAlmostEqual(total, cost - salvage, places=1)

View File

@@ -0,0 +1,59 @@
from datetime import date
from odoo.tests.common import TransactionCase
from odoo.tests import tagged
from odoo.exceptions import ValidationError
@tagged('post_install', '-at_install')
class TestFusionAsset(TransactionCase):
def setUp(self):
super().setUp()
self.asset_vals = {
'name': 'Test Asset',
'cost': 10000,
'salvage_value': 1000,
'acquisition_date': date(2026, 1, 1),
'method': 'straight_line',
'useful_life_years': 5,
}
def test_create_minimal(self):
a = self.env['fusion.asset'].create(self.asset_vals)
self.assertEqual(a.state, 'draft')
self.assertEqual(a.book_value, 10000)
def test_state_transitions_draft_to_running(self):
a = self.env['fusion.asset'].create(self.asset_vals)
a.action_set_running()
self.assertEqual(a.state, 'running')
self.assertTrue(a.in_service_date)
def test_pause_resume(self):
a = self.env['fusion.asset'].create(self.asset_vals)
a.action_set_running()
a.action_pause()
self.assertEqual(a.state, 'paused')
a.action_resume()
self.assertEqual(a.state, 'running')
def test_cannot_pause_from_draft(self):
a = self.env['fusion.asset'].create(self.asset_vals)
with self.assertRaises(ValidationError):
a.action_pause()
def test_negative_cost_rejected(self):
with self.assertRaises(Exception):
self.env['fusion.asset'].create({**self.asset_vals, 'cost': -100})
def test_salvage_exceeds_cost_rejected(self):
with self.assertRaises(Exception):
self.env['fusion.asset'].create(
{**self.asset_vals, 'cost': 1000, 'salvage_value': 5000},
)
def test_book_value_starts_at_cost(self):
a = self.env['fusion.asset'].create(self.asset_vals)
self.assertEqual(a.book_value, a.cost)
self.assertEqual(a.total_depreciated, 0)

View File

@@ -0,0 +1,49 @@
from datetime import date
from odoo.tests.common import TransactionCase
from odoo.tests import tagged
@tagged('post_install', '-at_install')
class TestFusionAssetAnomaly(TransactionCase):
def setUp(self):
super().setUp()
self.asset = self.env['fusion.asset'].create({
'name': 'Watched Asset',
'cost': 5000,
'acquisition_date': date(2026, 1, 1),
})
def _make_anomaly(self, **kw):
vals = {
'asset_id': self.asset.id,
'anomaly_type': 'behind_schedule',
'severity': 'medium',
'expected': 1000.0,
'actual': 700.0,
'variance_pct': -30.0,
}
vals.update(kw)
return self.env['fusion.asset.anomaly'].create(vals)
def test_create_defaults_state_new(self):
a = self._make_anomaly()
self.assertEqual(a.state, 'new')
self.assertTrue(a.detected_at)
self.assertEqual(a.company_id, self.asset.company_id)
def test_acknowledge_transitions(self):
a = self._make_anomaly()
a.action_acknowledge()
self.assertEqual(a.state, 'acknowledged')
def test_dismiss_transitions(self):
a = self._make_anomaly()
a.action_dismiss()
self.assertEqual(a.state, 'dismissed')
def test_resolve_transitions(self):
a = self._make_anomaly(anomaly_type='low_utilization', severity='high')
a.action_resolve()
self.assertEqual(a.state, 'resolved')

View File

@@ -0,0 +1,35 @@
from datetime import date
from odoo.tests.common import TransactionCase
from odoo.tests import tagged
@tagged('post_install', '-at_install')
class TestFusionAssetCategory(TransactionCase):
def test_create_with_defaults(self):
cat = self.env['fusion.asset.category'].create({'name': 'Computers'})
self.assertEqual(cat.method, 'straight_line')
self.assertEqual(cat.useful_life_years, 5)
self.assertEqual(cat.prorate_convention, 'days_period')
self.assertEqual(cat.asset_count, 0)
def test_asset_count_reflects_linked_assets(self):
cat = self.env['fusion.asset.category'].create({'name': 'Vehicles'})
for i in range(3):
self.env['fusion.asset'].create({
'name': f'Truck {i}',
'cost': 50000,
'acquisition_date': date(2026, 1, 1),
'method': 'declining_balance',
'category_id': cat.id,
})
cat.invalidate_recordset(['asset_count'])
self.assertEqual(cat.asset_count, 3)
def test_method_must_be_in_selection(self):
with self.assertRaises(Exception):
self.env['fusion.asset.category'].create({
'name': 'Bogus',
'method': 'not_a_method',
})

View File

@@ -0,0 +1,62 @@
from datetime import date
from odoo.tests.common import TransactionCase
from odoo.tests import tagged
@tagged('post_install', '-at_install')
class TestFusionAssetDepreciationLine(TransactionCase):
def setUp(self):
super().setUp()
self.asset = self.env['fusion.asset'].create({
'name': 'Asset for Lines',
'cost': 12000,
'salvage_value': 0,
'acquisition_date': date(2026, 1, 1),
'method': 'straight_line',
'useful_life_years': 1,
})
def _make_line(self, period_index, amount=1000.0, scheduled_date=None):
return self.env['fusion.asset.depreciation.line'].create({
'asset_id': self.asset.id,
'period_index': period_index,
'scheduled_date': scheduled_date or date(2026, period_index, 28),
'amount': amount,
})
def test_create_line_defaults_unposted(self):
line = self._make_line(1)
self.assertFalse(line.is_posted)
self.assertFalse(line.posted_date)
self.assertFalse(line.move_id)
self.assertEqual(line.company_id, self.asset.company_id)
self.assertEqual(line.currency_id, self.asset.currency_id)
def test_action_post_marks_line_posted(self):
line = self._make_line(2)
line.action_post()
self.assertTrue(line.is_posted)
self.assertTrue(line.posted_date)
def test_action_post_idempotent_keeps_first_date(self):
line = self._make_line(3)
line.action_post()
first_date = line.posted_date
line.action_post()
self.assertEqual(line.posted_date, first_date)
def test_unique_period_per_asset(self):
self._make_line(4)
with self.assertRaises(Exception):
self._make_line(4)
def test_book_value_reflects_posted_lines_only(self):
l1 = self._make_line(5, amount=1000)
self._make_line(6, amount=1500)
self.assertEqual(self.asset.book_value, 12000)
l1.action_post()
self.asset.invalidate_recordset(['book_value', 'total_depreciated'])
self.assertEqual(self.asset.total_depreciated, 1000)
self.assertEqual(self.asset.book_value, 11000)

View File

@@ -0,0 +1,56 @@
from datetime import date
from odoo.tests.common import TransactionCase
from odoo.tests import tagged
@tagged('post_install', '-at_install')
class TestFusionAssetDisposal(TransactionCase):
def setUp(self):
super().setUp()
self.asset = self.env['fusion.asset'].create({
'name': 'Disposable Asset',
'cost': 10000,
'salvage_value': 0,
'acquisition_date': date(2026, 1, 1),
'method': 'straight_line',
'useful_life_years': 5,
})
def test_create_minimal_sale(self):
d = self.env['fusion.asset.disposal'].create({
'asset_id': self.asset.id,
'disposal_type': 'sale',
'sale_amount': 7000,
'book_value_at_disposal': 6000,
})
self.assertEqual(d.gain_loss_amount, 1000)
self.assertEqual(d.company_id, self.asset.company_id)
def test_sale_at_loss(self):
d = self.env['fusion.asset.disposal'].create({
'asset_id': self.asset.id,
'disposal_type': 'sale',
'sale_amount': 4000,
'book_value_at_disposal': 6000,
})
self.assertEqual(d.gain_loss_amount, -2000)
def test_scrap_full_loss(self):
d = self.env['fusion.asset.disposal'].create({
'asset_id': self.asset.id,
'disposal_type': 'scrap',
'sale_amount': 0,
'book_value_at_disposal': 6000,
})
self.assertEqual(d.gain_loss_amount, -6000)
def test_donation_ignores_sale_amount(self):
d = self.env['fusion.asset.disposal'].create({
'asset_id': self.asset.id,
'disposal_type': 'donation',
'sale_amount': 999,
'book_value_at_disposal': 6000,
})
self.assertEqual(d.gain_loss_amount, -6000)

View File

@@ -0,0 +1,115 @@
from datetime import date
from odoo.tests.common import TransactionCase
from odoo.tests import tagged
from odoo.exceptions import ValidationError
@tagged('post_install', '-at_install')
class TestFusionAssetEngine(TransactionCase):
def setUp(self):
super().setUp()
self.engine = self.env['fusion.asset.engine']
self.asset = self.env['fusion.asset'].create({
'name': 'Test Engine Asset',
'cost': 10000,
'salvage_value': 1000,
'acquisition_date': date(2026, 1, 1),
'in_service_date': date(2026, 1, 1),
'method': 'straight_line',
'useful_life_years': 5,
})
def test_engine_model_exists(self):
self.assertIn('fusion.asset.engine', self.env.registry)
def test_compute_schedule_straight_line(self):
result = self.engine.compute_depreciation_schedule(self.asset)
self.assertEqual(result['lines_created'], 5)
lines = self.asset.depreciation_line_ids
self.assertEqual(len(lines), 5)
# Total depreciation should equal cost - salvage = 9000
total = sum(lines.mapped('amount'))
self.assertAlmostEqual(total, 9000, places=2)
def test_compute_schedule_declining_balance(self):
self.asset.write({'method': 'declining_balance', 'declining_rate_pct': 30.0})
self.engine.compute_depreciation_schedule(self.asset)
lines = self.asset.depreciation_line_ids
self.assertGreater(len(lines), 0)
# First-period amount should be cost * rate = 10000 * 0.3 = 3000
first = lines.sorted('period_index')[0]
self.assertAlmostEqual(first.amount, 3000, places=2)
def test_compute_schedule_recompute_wipes_unposted(self):
self.engine.compute_depreciation_schedule(self.asset)
self.asset.write({'useful_life_years': 8})
self.engine.compute_depreciation_schedule(self.asset, recompute=True)
self.assertEqual(len(self.asset.depreciation_line_ids), 8)
def test_compute_schedule_validates_zero_cost(self):
# Bypass DB constraint with sudo + the constraint allows cost >= 0,
# but engine validation requires cost > 0.
bad = self.env['fusion.asset'].create({
'name': 'Zero',
'cost': 0,
'acquisition_date': date(2026, 1, 1),
'method': 'straight_line',
'useful_life_years': 5,
})
with self.assertRaises(ValidationError):
self.engine.compute_depreciation_schedule(bad)
def test_post_depreciation_entry_marks_line_posted(self):
self.engine.compute_depreciation_schedule(self.asset)
self.asset.action_set_running()
result = self.engine.post_depreciation_entry(self.asset)
self.assertEqual(result['posted_count'], 1)
first_line = self.asset.depreciation_line_ids.sorted('period_index')[0]
self.assertTrue(first_line.is_posted)
def test_post_depreciation_only_after_running(self):
self.engine.compute_depreciation_schedule(self.asset)
# asset is still in 'draft' state
with self.assertRaises(ValidationError):
self.engine.post_depreciation_entry(self.asset)
def test_dispose_asset_creates_disposal_record(self):
self.engine.compute_depreciation_schedule(self.asset)
self.asset.action_set_running()
result = self.engine.dispose_asset(
self.asset, sale_amount=5000, sale_date=date(2027, 6, 1),
)
self.assertEqual(self.asset.state, 'disposed')
self.assertIn('disposal_id', result)
self.assertEqual(result['book_value_at_disposal'], self.asset.book_value)
def test_partial_sale_creates_child_and_disposes(self):
self.engine.compute_depreciation_schedule(self.asset)
self.asset.action_set_running()
original_cost = self.asset.cost
result = self.engine.partial_sale(
self.asset, sold_amount=3000, sold_qty=0.3,
sale_date=date(2027, 6, 1),
)
self.assertIn('parent_asset_id', result)
self.assertIn('child_asset_id', result)
self.asset.invalidate_recordset(['cost'])
expected_remaining = round(original_cost * 0.7, 2)
self.assertAlmostEqual(self.asset.cost, expected_remaining, places=2)
def test_pause_resume_round_trip(self):
self.asset.action_set_running()
self.engine.pause_asset(self.asset)
self.assertEqual(self.asset.state, 'paused')
self.engine.resume_asset(self.asset)
self.assertEqual(self.asset.state, 'running')
def test_reverse_disposal_restores_running_state(self):
self.engine.compute_depreciation_schedule(self.asset)
self.asset.action_set_running()
self.engine.dispose_asset(self.asset, sale_amount=5000)
self.assertEqual(self.asset.state, 'disposed')
self.engine.reverse_disposal(self.asset)
self.assertEqual(self.asset.state, 'running')

View File

@@ -0,0 +1,83 @@
"""Local LLM compat smoke test for the useful_life_predictor service.
Auto-detects an LM Studio (port 1234) or Ollama (port 11434) server on
host.docker.internal or localhost. Skips silently when no local LLM is
reachable, so CI runs stay green.
When a server is present, this exercises the real OpenAI-compatible
adapter end-to-end against a local model — i.e. it catches prompt /
JSON-parsing regressions that only show up with a non-mocked LLM.
"""
import socket
from odoo.tests.common import TransactionCase
from odoo.tests import tagged
def _server_reachable(host, port, timeout=1.0):
try:
with socket.create_connection((host, port), timeout=timeout):
return True
except (OSError, socket.timeout):
return False
def _detect_local_llm():
candidates = [
('host.docker.internal', 1234, 'local-model'),
('host.docker.internal', 11434, 'llama3.1:8b'),
('localhost', 1234, 'local-model'),
('localhost', 11434, 'llama3.1:8b'),
]
for host, port, default_model in candidates:
if _server_reachable(host, port, timeout=0.5):
return (f'http://{host}:{port}/v1', default_model)
return (None, None)
@tagged('post_install', '-at_install', 'local_llm')
class TestLocalLLMUsefulLife(TransactionCase):
def setUp(self):
super().setUp()
self.base_url, self.model = _detect_local_llm()
if not self.base_url:
self.skipTest("No local LLM server detected (LM Studio :1234 / Ollama :11434)")
def test_useful_life_with_local_llm(self):
params = self.env['ir.config_parameter'].sudo()
keys = [
'fusion_accounting.openai_base_url',
'fusion_accounting.openai_model',
'fusion_accounting.openai_api_key',
'fusion_accounting.provider.asset_useful_life',
]
prior = {k: params.get_param(k) for k in keys}
params.set_param('fusion_accounting.openai_base_url', self.base_url)
params.set_param('fusion_accounting.openai_model', self.model)
params.set_param('fusion_accounting.openai_api_key', 'lm-studio')
params.set_param('fusion_accounting.provider.asset_useful_life', 'openai')
try:
from odoo.addons.fusion_accounting_assets.services.useful_life_predictor import (
predict_useful_life,
)
result = predict_useful_life(
self.env,
description='Dell laptop',
amount=2500,
partner_name='Dell Canada',
)
self.assertIn('useful_life_years', result)
self.assertIn('depreciation_method', result)
self.assertIsInstance(result['useful_life_years'], (int, float))
self.assertIn(
result['depreciation_method'],
('straight_line', 'declining_balance', 'units_of_production'),
)
finally:
for k, v in prior.items():
if v is not None:
params.set_param(k, v)

View File

@@ -0,0 +1,112 @@
"""Integration tests verifying all 3 depreciation methods through the engine."""
from datetime import date
from odoo.tests.common import TransactionCase
from odoo.tests import tagged
@tagged('post_install', '-at_install', 'integration')
class TestStraightLineIntegration(TransactionCase):
def setUp(self):
super().setUp()
self.engine = self.env['fusion.asset.engine']
def test_straight_line_5yr_no_salvage(self):
asset = self.env['fusion.asset'].create({
'name': 'SL Test', 'cost': 10000, 'salvage_value': 0,
'acquisition_date': date(2026, 1, 1),
'in_service_date': date(2026, 1, 1),
'method': 'straight_line', 'useful_life_years': 5,
})
self.engine.compute_depreciation_schedule(asset)
lines = asset.depreciation_line_ids.sorted('period_index')
self.assertEqual(len(lines), 5)
for line in lines:
self.assertAlmostEqual(line.amount, 2000, places=2)
def test_straight_line_10yr_with_salvage(self):
asset = self.env['fusion.asset'].create({
'name': 'SL10', 'cost': 50000, 'salvage_value': 5000,
'acquisition_date': date(2026, 1, 1),
'in_service_date': date(2026, 1, 1),
'method': 'straight_line', 'useful_life_years': 10,
})
self.engine.compute_depreciation_schedule(asset)
lines = asset.depreciation_line_ids.sorted('period_index')
self.assertEqual(len(lines), 10)
# Each year = (50000-5000)/10 = 4500; total depreciable = 45000
self.assertAlmostEqual(sum(lines.mapped('amount')), 45000, places=2)
def test_straight_line_book_value_at_end_equals_salvage(self):
asset = self.env['fusion.asset'].create({
'name': 'SL', 'cost': 10000, 'salvage_value': 1000,
'acquisition_date': date(2026, 1, 1),
'in_service_date': date(2026, 1, 1),
'method': 'straight_line', 'useful_life_years': 5,
})
self.engine.compute_depreciation_schedule(asset)
last = asset.depreciation_line_ids.sorted('period_index')[-1]
self.assertAlmostEqual(last.book_value_at_end, 1000, places=2)
@tagged('post_install', '-at_install', 'integration')
class TestDecliningBalanceIntegration(TransactionCase):
def setUp(self):
super().setUp()
self.engine = self.env['fusion.asset.engine']
def test_declining_balance_30pct(self):
asset = self.env['fusion.asset'].create({
'name': 'DB', 'cost': 10000, 'salvage_value': 1000,
'acquisition_date': date(2026, 1, 1),
'in_service_date': date(2026, 1, 1),
'method': 'declining_balance', 'useful_life_years': 5,
'declining_rate_pct': 30.0,
})
self.engine.compute_depreciation_schedule(asset)
lines = asset.depreciation_line_ids.sorted('period_index')
# First period: 10000 * 0.30 = 3000
self.assertAlmostEqual(lines[0].amount, 3000, places=2)
# Should not exceed salvage at end
self.assertGreaterEqual(lines[-1].book_value_at_end, 999.99)
def test_declining_balance_50pct_high_rate(self):
asset = self.env['fusion.asset'].create({
'name': 'DB50', 'cost': 8000, 'salvage_value': 500,
'acquisition_date': date(2026, 1, 1),
'in_service_date': date(2026, 1, 1),
'method': 'declining_balance', 'useful_life_years': 5,
'declining_rate_pct': 50.0,
})
self.engine.compute_depreciation_schedule(asset)
# First period: 8000 * 0.50 = 4000
first = asset.depreciation_line_ids.sorted('period_index')[0]
self.assertAlmostEqual(first.amount, 4000, places=2)
@tagged('post_install', '-at_install', 'integration')
class TestUnitsOfProductionIntegration(TransactionCase):
def setUp(self):
super().setUp()
self.engine = self.env['fusion.asset.engine']
def test_units_of_production_5yr_even_distribution(self):
asset = self.env['fusion.asset'].create({
'name': 'UOP', 'cost': 50000, 'salvage_value': 0,
'acquisition_date': date(2026, 1, 1),
'in_service_date': date(2026, 1, 1),
'method': 'units_of_production',
'total_units_expected': 100000,
'useful_life_years': 5,
})
self.engine.compute_depreciation_schedule(asset)
lines = asset.depreciation_line_ids.sorted('period_index')
# 5 periods, even distribution = 20000 units/period
# Each period: (20000/100000) * 50000 = 10000
self.assertEqual(len(lines), 5)
for line in lines:
self.assertAlmostEqual(line.amount, 10000, places=2)

View File

@@ -0,0 +1,24 @@
from odoo.tests.common import TransactionCase
from odoo.tests import tagged
@tagged('post_install', '-at_install')
class TestAssetsMigrationRoundTrip(TransactionCase):
def test_bootstrap_step_runs_without_enterprise(self):
"""When Enterprise account.asset is NOT installed, step is a no-op."""
wizard = self.env['fusion.migration.wizard'].create({})
result = wizard._assets_bootstrap_step()
self.assertEqual(result['step'], 'assets_bootstrap')
# In our local DB, Enterprise account.asset may or may not exist
# If absent: enterprise_module_present is False
# If present: created>=0
self.assertIn(result['enterprise_module_present'], [True, False])
def test_bootstrap_idempotent_on_re_run(self):
wizard = self.env['fusion.migration.wizard'].create({})
first = wizard._assets_bootstrap_step()
second = wizard._assets_bootstrap_step()
# Second run should skip what the first created (or both no-op)
if first['enterprise_module_present']:
self.assertGreaterEqual(second['skipped'], first['created'])

View File

@@ -0,0 +1,48 @@
from datetime import date
from odoo.exceptions import UserError
from odoo.tests import tagged
from odoo.tests.common import TransactionCase
@tagged('post_install', '-at_install')
class TestPartialSaleWizard(TransactionCase):
def setUp(self):
super().setUp()
self.asset = self.env['fusion.asset'].create({
'name': 'Partial Sale Test',
'cost': 10000,
'acquisition_date': date(2026, 1, 1),
'in_service_date': date(2026, 1, 1),
'method': 'straight_line', 'useful_life_years': 5,
})
self.env['fusion.asset.engine'].compute_depreciation_schedule(self.asset)
self.asset.action_set_running()
def test_partial_sell_30pct_creates_child(self):
wizard = self.env['fusion.partial.sale.wizard'].create({
'asset_id': self.asset.id,
'sold_pct': 30.0, 'sold_amount': 4000,
'sale_date': date(2026, 6, 1),
})
wizard.action_partial_sell()
self.asset.invalidate_recordset(['cost'])
self.assertAlmostEqual(self.asset.cost, 7000, places=2)
def test_invalid_pct_raises(self):
wizard = self.env['fusion.partial.sale.wizard'].create({
'asset_id': self.asset.id,
'sold_pct': 0, 'sold_amount': 100,
})
with self.assertRaises(UserError):
wizard.action_partial_sell()
def test_compute_estimated_gain_loss(self):
wizard = self.env['fusion.partial.sale.wizard'].new({
'asset_id': self.asset.id,
'sold_pct': 30.0, 'sold_amount': 4000,
})
wizard._compute_sold_cost()
self.assertAlmostEqual(wizard.estimated_sold_cost, 3000, places=2)
self.assertAlmostEqual(wizard.estimated_gain_loss, 1000, places=2)

View File

@@ -0,0 +1,58 @@
"""Controller perf benchmarks tagged 'benchmark'.
Engine-level benchmarks live in test_performance_benchmarks.py (Task 23).
This file targets the JSON-RPC controller surface end-to-end (HTTP request
→ Odoo dispatch → engine → response). It complements Task 23 by catching
regressions introduced by controller / serialization layers, not just the
underlying engine.
"""
import json
import statistics
import time
from datetime import date
from odoo.tests.common import HttpCase, new_test_user
from odoo.tests import tagged
@tagged('post_install', '-at_install', 'benchmark')
class TestAssetsControllerBenchmarks(HttpCase):
def setUp(self):
super().setUp()
for i in range(15):
self.env['fusion.asset'].create({
'name': f'BenchAsset{i}',
'cost': 1000 + i * 100,
'acquisition_date': date(2026, 1, 1),
'method': 'straight_line',
'useful_life_years': 5,
})
def test_get_detail_endpoint_p95(self):
new_test_user(
self.env, login='asset_perf_ctrl',
groups='base.group_user,account.group_account_invoice',
)
asset = self.env['fusion.asset'].search([], limit=1)
self.authenticate('asset_perf_ctrl', 'asset_perf_ctrl')
timings = []
for _ in range(5):
start = time.perf_counter()
response = self.url_open(
'/fusion/assets/get_detail',
data=json.dumps({
'jsonrpc': '2.0', 'method': 'call', 'id': 1,
'params': {'asset_id': asset.id},
}),
headers={'Content-Type': 'application/json'},
)
timings.append((time.perf_counter() - start) * 1000)
self.assertEqual(response.status_code, 200)
sorted_t = sorted(timings)
p95 = sorted_t[min(int(len(sorted_t) * 0.95), len(sorted_t) - 1)]
median = statistics.median(timings)
msg = f"controller.get_detail: median={median:.0f}ms p95={p95:.0f}ms"
print(f"\n PERF: {msg} (target <500ms)")
self.assertLess(p95, 5000)

View File

@@ -0,0 +1,117 @@
"""Performance benchmarks tagged 'benchmark'."""
import json
import statistics
import time
from datetime import date
from odoo.tests.common import HttpCase, TransactionCase, new_test_user
from odoo.tests import tagged
@tagged('post_install', '-at_install', 'benchmark')
class TestEngineBenchmarks(TransactionCase):
def setUp(self):
super().setUp()
self.engine = self.env['fusion.asset.engine']
def _percentile(self, samples, p):
if len(samples) <= 1:
return samples[0] if samples else 0
sorted_s = sorted(samples)
idx = int(len(sorted_s) * p / 100)
return sorted_s[min(idx, len(sorted_s) - 1)]
def test_compute_schedule_p95(self):
timings = []
for i in range(10):
asset = self.env['fusion.asset'].create({
'name': f'PerfAsset{i}', 'cost': 100000, 'salvage_value': 5000,
'acquisition_date': date(2026, 1, 1),
'method': 'straight_line', 'useful_life_years': 10,
})
start = time.perf_counter()
self.engine.compute_depreciation_schedule(asset)
timings.append((time.perf_counter() - start) * 1000)
p95 = self._percentile(timings, 95)
median = statistics.median(timings)
msg = f"compute_schedule(10yr): median={median:.0f}ms p95={p95:.0f}ms"
print(f"\n PERF: {msg} (target <500ms)")
self.assertLess(p95, 5000, f"way over budget: {msg}")
def test_post_depreciation_p95(self):
asset = self.env['fusion.asset'].create({
'name': 'PostPerf', 'cost': 50000,
'acquisition_date': date(2026, 1, 1),
'in_service_date': date(2026, 1, 1),
'method': 'straight_line', 'useful_life_years': 10,
})
self.engine.compute_depreciation_schedule(asset)
asset.action_set_running()
timings = []
for _ in range(5):
start = time.perf_counter()
self.engine.post_depreciation_entry(asset)
timings.append((time.perf_counter() - start) * 1000)
p95 = self._percentile(timings, 95)
median = statistics.median(timings)
msg = f"post_depreciation: median={median:.0f}ms p95={p95:.0f}ms"
print(f"\n PERF: {msg} (target <300ms)")
self.assertLess(p95, 3000)
def test_dispose_asset_p95(self):
timings = []
for i in range(5):
asset = self.env['fusion.asset'].create({
'name': f'DispPerf{i}', 'cost': 10000,
'acquisition_date': date(2026, 1, 1),
'in_service_date': date(2026, 1, 1),
'method': 'straight_line', 'useful_life_years': 5,
})
self.engine.compute_depreciation_schedule(asset)
asset.action_set_running()
start = time.perf_counter()
self.engine.dispose_asset(asset, sale_amount=5000)
timings.append((time.perf_counter() - start) * 1000)
p95 = self._percentile(timings, 95)
median = statistics.median(timings)
msg = f"dispose_asset: median={median:.0f}ms p95={p95:.0f}ms"
print(f"\n PERF: {msg} (target <300ms)")
self.assertLess(p95, 3000)
@tagged('post_install', '-at_install', 'benchmark')
class TestControllerBenchmarks(HttpCase):
def test_list_endpoint_p95(self):
new_test_user(
self.env, login='asset_perf',
groups='base.group_user,account.group_account_invoice',
)
for i in range(20):
self.env['fusion.asset'].create({
'name': f'ListPerf{i}', 'cost': 1000,
'acquisition_date': date(2026, 1, 1),
'method': 'straight_line', 'useful_life_years': 4,
})
self.authenticate('asset_perf', 'asset_perf')
timings = []
for _ in range(5):
start = time.perf_counter()
response = self.url_open(
'/fusion/assets/list',
data=json.dumps({
'jsonrpc': '2.0', 'method': 'call', 'id': 1,
'params': {'company_id': self.env.company.id},
}),
headers={'Content-Type': 'application/json'},
)
timings.append((time.perf_counter() - start) * 1000)
self.assertEqual(response.status_code, 200)
sorted_t = sorted(timings)
p95 = sorted_t[min(int(len(sorted_t) * 0.95), len(sorted_t) - 1)]
median = statistics.median(timings)
msg = f"controller.list: median={median:.0f}ms p95={p95:.0f}ms"
print(f"\n PERF: {msg} (target <300ms)")
self.assertLess(p95, 3000)

View File

@@ -0,0 +1,65 @@
from datetime import date
from odoo.tests.common import TransactionCase
from odoo.tests import tagged
from odoo.addons.fusion_accounting_assets.services.prorate import prorate_factor
@tagged('post_install', '-at_install')
class TestProrate(TransactionCase):
def test_full_month_convention_always_one(self):
f = prorate_factor(
period_start=date(2026, 1, 1),
period_end=date(2026, 1, 31),
asset_start=date(2026, 1, 15),
convention='full_month',
)
self.assertEqual(f, 1.0)
def test_asset_starts_before_period_full_factor(self):
f = prorate_factor(
period_start=date(2026, 1, 1),
period_end=date(2026, 1, 31),
asset_start=date(2025, 12, 1),
convention='days_period',
)
self.assertEqual(f, 1.0)
def test_asset_starts_after_period_zero_factor(self):
f = prorate_factor(
period_start=date(2026, 1, 1),
period_end=date(2026, 1, 31),
asset_start=date(2026, 2, 5),
convention='days_period',
)
self.assertEqual(f, 0.0)
def test_days_period_mid_month(self):
# Jan 16 -> Jan 31 inclusive = 16 days; period = 31 days
f = prorate_factor(
period_start=date(2026, 1, 1),
period_end=date(2026, 1, 31),
asset_start=date(2026, 1, 16),
convention='days_period',
)
self.assertAlmostEqual(f, 16 / 31, places=5)
def test_days_365_mid_month(self):
# 16 days / 365
f = prorate_factor(
period_start=date(2026, 1, 1),
period_end=date(2026, 1, 31),
asset_start=date(2026, 1, 16),
convention='days_365',
)
self.assertAlmostEqual(f, 16 / 365.0, places=5)
def test_unknown_convention_raises(self):
with self.assertRaises(ValueError):
prorate_factor(
period_start=date(2026, 1, 1),
period_end=date(2026, 1, 31),
asset_start=date(2026, 1, 15),
convention='bogus', # type: ignore[arg-type]
)

View File

@@ -0,0 +1,45 @@
from odoo.tests.common import TransactionCase
from odoo.tests import tagged
from odoo.addons.fusion_accounting_assets.services.salvage_value import (
SalvageConfig, compute_salvage_value, remaining_useful_life_value,
)
@tagged('post_install', '-at_install')
class TestSalvageValue(TransactionCase):
def test_zero_method_returns_zero(self):
v = compute_salvage_value(cost=10000, config=SalvageConfig(method='zero'))
self.assertEqual(v, 0.0)
def test_percentage_method(self):
v = compute_salvage_value(
cost=10000, config=SalvageConfig(method='percentage', value=10),
)
self.assertAlmostEqual(v, 1000.0, places=2)
def test_fixed_method(self):
v = compute_salvage_value(
cost=10000, config=SalvageConfig(method='fixed', value=750),
)
self.assertAlmostEqual(v, 750.0, places=2)
def test_unknown_method_raises(self):
with self.assertRaises(ValueError):
compute_salvage_value(
cost=10000,
config=SalvageConfig(method='bogus', value=0), # type: ignore[arg-type]
)
def test_remaining_useful_life_value_midway(self):
# Halfway through life; current book 6000, salvage 1000 -> 1000 + 5000*0.5 = 3500
v = remaining_useful_life_value(
current_book=6000, salvage=1000, periods_used=5, total_periods=10,
)
self.assertAlmostEqual(v, 3500.0, places=2)
def test_remaining_useful_life_value_at_end_returns_salvage(self):
v = remaining_useful_life_value(
current_book=1200, salvage=1000, periods_used=10, total_periods=10,
)
self.assertEqual(v, 1000.0)

View File

@@ -0,0 +1,61 @@
from odoo.tests.common import TransactionCase
from odoo.tests import tagged
from odoo.addons.fusion_accounting_assets.services.useful_life_predictor import (
predict_useful_life,
)
from odoo.addons.fusion_accounting_assets.services.useful_life_prompt import (
SYSTEM_PROMPT, build_prompt,
)
@tagged('post_install', '-at_install')
class TestUsefulLifePredictor(TransactionCase):
def setUp(self):
super().setUp()
# Ensure no provider configured for these fallback tests.
self.env['ir.config_parameter'].sudo().search([
('key', 'in', [
'fusion_accounting.provider.asset_useful_life',
'fusion_accounting.provider.default',
])
]).unlink()
def test_fallback_computer(self):
result = predict_useful_life(self.env, description="Dell laptop")
self.assertEqual(result['useful_life_years'], 4)
self.assertEqual(result['depreciation_method'], 'straight_line')
def test_fallback_furniture(self):
result = predict_useful_life(self.env, description="office desk")
self.assertEqual(result['useful_life_years'], 7)
def test_fallback_vehicle_uses_declining(self):
result = predict_useful_life(self.env, description="Ford F-150 truck")
self.assertEqual(result['useful_life_years'], 5)
self.assertEqual(result['depreciation_method'], 'declining_balance')
def test_fallback_default_for_unknown(self):
result = predict_useful_life(self.env, description="mystery widget")
self.assertEqual(result['useful_life_years'], 5)
self.assertEqual(result['confidence'], 0.3)
def test_returns_dict_with_required_keys(self):
result = predict_useful_life(self.env, description="server")
for key in ('useful_life_years', 'depreciation_method', 'rationale', 'confidence'):
self.assertIn(key, result)
@tagged('post_install', '-at_install')
class TestUsefulLifePrompt(TransactionCase):
def test_system_prompt_requires_json(self):
self.assertIn('JSON', SYSTEM_PROMPT)
def test_build_prompt_returns_tuple(self):
result = build_prompt(description='test')
self.assertEqual(len(result), 2)
def test_user_prompt_includes_amount(self):
_, user = build_prompt(description='laptop', amount=2000)
self.assertIn('2,000', user)