91 lines
3.8 KiB
Python
91 lines
3.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
from odoo.tests.common import TransactionCase
|
|
|
|
class TestMetaCatalogSync(TransactionCase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.company = self.env.company
|
|
self.company.write({
|
|
'meta_sync_enabled': True,
|
|
'meta_catalog_id': 'test_catalog_12345',
|
|
'meta_access_token': 'test_token_secret_xyz',
|
|
'meta_auto_sync_on_write': True,
|
|
})
|
|
self.category = self.env['product.category'].create({'name': 'Dine360 Specials'})
|
|
|
|
self.template = self.env['product.template'].create({
|
|
'name': 'Dine360 Signature Burger',
|
|
'list_price': 14.50,
|
|
'categ_id': self.category.id,
|
|
'is_published': True,
|
|
'meta_sync_enabled': True,
|
|
'detailed_type': 'consu',
|
|
})
|
|
|
|
def test_01_build_item_payload(self):
|
|
"""Verify Meta catalog payload generation contains all required commerce fields."""
|
|
variant = self.template.product_variant_ids[:1]
|
|
variant.default_code = "BURGER-001"
|
|
|
|
meta_api = self.env['social.meta.catalog']
|
|
payload = meta_api.build_item_payload(variant, method='UPDATE')
|
|
|
|
self.assertEqual(payload['method'], 'UPDATE')
|
|
self.assertEqual(payload['retailer_id'], 'BURGER-001')
|
|
|
|
data = payload['data']
|
|
self.assertEqual(data['title'], 'Dine360 Signature Burger')
|
|
self.assertEqual(data['availability'], 'in stock')
|
|
self.assertEqual(data['price'], 1450) # 14.50 * 100
|
|
self.assertEqual(data['currency'], self.company.currency_id.name)
|
|
self.assertIn('/shop', data['link'])
|
|
self.assertIn(f'/web/image/product.product/{variant.id}/image_1920', data['image_link'])
|
|
self.assertEqual(data['item_group_id'], f"odoo_tmpl_{self.template.id}")
|
|
|
|
def test_02_product_write_triggers_queue(self):
|
|
"""Verify modifying price enqueues the product into social.meta.sync.queue."""
|
|
variant = self.template.product_variant_ids[:1]
|
|
|
|
# Trigger price change
|
|
self.template.write({'list_price': 16.99})
|
|
|
|
queue_item = self.env['social.meta.sync.queue'].search([
|
|
('product_id', '=', variant.id),
|
|
('state', 'in', ('pending', 'processing', 'done'))
|
|
], limit=1)
|
|
|
|
self.assertTrue(queue_item, "A sync queue item should be created when product price changes.")
|
|
self.assertEqual(queue_item.operation, 'create_update')
|
|
|
|
def test_03_unpublish_triggers_delete_operation(self):
|
|
"""Verify unpublishing a product flags it for deletion in Meta Catalog."""
|
|
variant = self.template.product_variant_ids[:1]
|
|
|
|
self.template.write({'is_published': False})
|
|
|
|
queue_item = self.env['social.meta.sync.queue'].search([
|
|
('product_id', '=', variant.id),
|
|
('operation', '=', 'delete')
|
|
], limit=1)
|
|
|
|
self.assertTrue(queue_item, "Unpublishing product should enqueue a delete operation.")
|
|
|
|
def test_04_sync_queue_batch_processing_resilience(self):
|
|
"""Verify process_queue safely catches errors and records them without raising unhandled exceptions."""
|
|
queue_model = self.env['social.meta.sync.queue']
|
|
variant = self.template.product_variant_ids[:1]
|
|
|
|
# Clear company credentials to simulate API configuration failure
|
|
self.company.write({'meta_catalog_id': False})
|
|
|
|
queue_item = queue_model.enqueue_product(variant, operation='create_update')
|
|
self.assertTrue(queue_item)
|
|
|
|
# Process queue should not raise UserError or break transactions
|
|
processed = queue_model.process_queue(limit=10)
|
|
self.assertEqual(processed, 0)
|
|
queue_item.invalidate_recordset()
|
|
self.assertEqual(queue_item.state, 'failed')
|
|
self.assertTrue(queue_item.last_error)
|