Compare commits
9 commits
Author | SHA1 | Date | |
---|---|---|---|
b9a34003c1 | |||
ff573cc17d | |||
a0b36a6384 | |||
bad6ded72d | |||
198c3f77bb | |||
c9d093d343 | |||
5d3820a8f2 | |||
![]() |
7721ddf11e | ||
![]() |
ebb9211f2d |
18 changed files with 415 additions and 245 deletions
4
.pylintrc
Normal file
4
.pylintrc
Normal file
|
@ -0,0 +1,4 @@
|
||||||
|
# -*- mode: conf; -*-
|
||||||
|
|
||||||
|
[MESSAGES CONTROL]
|
||||||
|
disable=fixme
|
|
@ -1,3 +1,9 @@
|
||||||
|
## v0.1.1 (2025-02-20)
|
||||||
|
|
||||||
|
### Fix
|
||||||
|
|
||||||
|
- set vendor name, sku when refreshing neworder batch row
|
||||||
|
|
||||||
## v0.1.0 (2025-01-13)
|
## v0.1.0 (2025-01-13)
|
||||||
|
|
||||||
### Feat
|
### Feat
|
||||||
|
|
|
@ -1,6 +1,7 @@
|
||||||
|
|
||||||
# Sideshow-COREPOS
|
# Sideshow-COREPOS
|
||||||
|
|
||||||
This package adds CORE-POS integration for Sideshow.
|
This package adds [CORE-POS](https://www.core-pos.com/) integration
|
||||||
|
for [Sideshow](https://pypi.org/project/Sideshow/).
|
||||||
|
|
||||||
Full docs are at https://rattailproject.org/docs/sideshow-corepos/
|
Full docs are at https://rattailproject.org/docs/sideshow-corepos/
|
||||||
|
|
29
docs/conf.py
29
docs/conf.py
|
@ -8,33 +8,32 @@
|
||||||
|
|
||||||
from importlib.metadata import version as get_version
|
from importlib.metadata import version as get_version
|
||||||
|
|
||||||
project = 'Sideshow-COREPOS'
|
project = "Sideshow-COREPOS"
|
||||||
copyright = '2025, Lance Edgar'
|
copyright = "2025, Lance Edgar"
|
||||||
author = 'Lance Edgar'
|
author = "Lance Edgar"
|
||||||
release = '0.1'
|
release = get_version("Sideshow-COREPOS")
|
||||||
release = get_version('Sideshow-COREPOS')
|
|
||||||
|
|
||||||
# -- General configuration ---------------------------------------------------
|
# -- General configuration ---------------------------------------------------
|
||||||
# https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration
|
# https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration
|
||||||
|
|
||||||
extensions = [
|
extensions = [
|
||||||
'sphinx.ext.autodoc',
|
"sphinx.ext.autodoc",
|
||||||
'sphinx.ext.intersphinx',
|
"sphinx.ext.intersphinx",
|
||||||
'sphinx.ext.viewcode',
|
"sphinx.ext.viewcode",
|
||||||
'sphinx.ext.todo',
|
"sphinx.ext.todo",
|
||||||
]
|
]
|
||||||
|
|
||||||
templates_path = ['_templates']
|
templates_path = ["_templates"]
|
||||||
exclude_patterns = ['_build', 'Thumbs.db', '.DS_Store']
|
exclude_patterns = ["_build", "Thumbs.db", ".DS_Store"]
|
||||||
|
|
||||||
intersphinx_mapping = {
|
intersphinx_mapping = {
|
||||||
'sideshow': ('https://rattailproject.org/docs/sideshow/', None),
|
"sideshow": ("https://docs.wuttaproject.org/sideshow/", None),
|
||||||
'wuttjamaican': ('https://rattailproject.org/docs/wuttjamaican/', None),
|
"wuttjamaican": ("https://docs.wuttaproject.org/wuttjamaican/", None),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
# -- Options for HTML output -------------------------------------------------
|
# -- Options for HTML output -------------------------------------------------
|
||||||
# https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output
|
# https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output
|
||||||
|
|
||||||
html_theme = 'furo'
|
html_theme = "furo"
|
||||||
html_static_path = ['_static']
|
html_static_path = ["_static"]
|
||||||
|
|
|
@ -8,6 +8,12 @@ This is `Sideshow`_ with integration for `CORE-POS`_.
|
||||||
|
|
||||||
.. _CORE-POS: https://www.core-pos.com/
|
.. _CORE-POS: https://www.core-pos.com/
|
||||||
|
|
||||||
|
.. image:: https://img.shields.io/badge/linting-pylint-yellowgreen
|
||||||
|
:target: https://github.com/pylint-dev/pylint
|
||||||
|
|
||||||
|
.. image:: https://img.shields.io/badge/code%20style-black-000000.svg
|
||||||
|
:target: https://github.com/psf/black
|
||||||
|
|
||||||
|
|
||||||
.. toctree::
|
.. toctree::
|
||||||
:maxdepth: 1
|
:maxdepth: 1
|
||||||
|
|
|
@ -5,7 +5,7 @@ build-backend = "hatchling.build"
|
||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "Sideshow-COREPOS"
|
name = "Sideshow-COREPOS"
|
||||||
version = "0.1.0"
|
version = "0.1.1"
|
||||||
description = "Case/Special Order Tracker for CORE-POS"
|
description = "Case/Special Order Tracker for CORE-POS"
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
authors = [
|
authors = [
|
||||||
|
@ -27,17 +27,19 @@ classifiers = [
|
||||||
"Programming Language :: Python :: 3.9",
|
"Programming Language :: Python :: 3.9",
|
||||||
"Programming Language :: Python :: 3.10",
|
"Programming Language :: Python :: 3.10",
|
||||||
"Programming Language :: Python :: 3.11",
|
"Programming Language :: Python :: 3.11",
|
||||||
|
"Topic :: Office/Business",
|
||||||
|
"Topic :: Software Development :: Libraries :: Python Modules",
|
||||||
]
|
]
|
||||||
license = {text = "GNU General Public License v3+"}
|
license = {text = "GNU General Public License v3+"}
|
||||||
requires-python = ">= 3.8"
|
requires-python = ">= 3.8"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"Sideshow>=0.3.0",
|
"Sideshow>=0.7.1",
|
||||||
"Wutta-COREPOS[web]>=0.2.0",
|
"Wutta-COREPOS[web]>=0.3.0",
|
||||||
]
|
]
|
||||||
|
|
||||||
[project.optional-dependencies]
|
[project.optional-dependencies]
|
||||||
docs = ["Sphinx", "furo"]
|
docs = ["Sphinx", "furo"]
|
||||||
tests = ["pytest-cov", "tox"]
|
tests = ["pylint", "pytest", "pytest-cov", "tox"]
|
||||||
|
|
||||||
|
|
||||||
[project.entry-points."paste.app_factory"]
|
[project.entry-points."paste.app_factory"]
|
||||||
|
|
|
@ -43,28 +43,35 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
|
||||||
for more info.
|
for more info.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def autocomplete_customers_external(self, session, term, user=None):
|
def autocomplete_customers_external( # pylint: disable=empty-docstring
|
||||||
|
self, session, term, user=None
|
||||||
|
):
|
||||||
""" """
|
""" """
|
||||||
corepos = self.app.get_corepos_handler()
|
corepos = self.app.get_corepos_handler()
|
||||||
op_model = corepos.get_model_office_op()
|
op_model = corepos.get_model_office_op()
|
||||||
op_session = corepos.make_session_office_op()
|
op_session = corepos.make_session_office_op()
|
||||||
|
|
||||||
# base query
|
# base query
|
||||||
query = op_session.query(op_model.CustomerClassic)\
|
query = op_session.query(op_model.CustomerClassic).join(
|
||||||
.join(op_model.MemberInfo,
|
op_model.MemberInfo,
|
||||||
op_model.MemberInfo.card_number == op_model.CustomerClassic.card_number)
|
op_model.MemberInfo.card_number == op_model.CustomerClassic.card_number,
|
||||||
|
)
|
||||||
|
|
||||||
# filter query
|
# filter query
|
||||||
criteria = []
|
criteria = []
|
||||||
for word in term.split():
|
for word in term.split():
|
||||||
criteria.append(sa.or_(
|
criteria.append(
|
||||||
op_model.CustomerClassic.first_name.ilike(f'%{word}%'),
|
sa.or_(
|
||||||
op_model.CustomerClassic.last_name.ilike(f'%{word}%')))
|
op_model.CustomerClassic.first_name.ilike(f"%{word}%"),
|
||||||
|
op_model.CustomerClassic.last_name.ilike(f"%{word}%"),
|
||||||
|
)
|
||||||
|
)
|
||||||
query = query.filter(sa.and_(*criteria))
|
query = query.filter(sa.and_(*criteria))
|
||||||
|
|
||||||
# sort query
|
# sort query
|
||||||
query = query.order_by(op_model.CustomerClassic.first_name,
|
query = query.order_by(
|
||||||
op_model.CustomerClassic.last_name)
|
op_model.CustomerClassic.first_name, op_model.CustomerClassic.last_name
|
||||||
|
)
|
||||||
|
|
||||||
# get data
|
# get data
|
||||||
# TODO: need max_results option
|
# TODO: need max_results option
|
||||||
|
@ -72,32 +79,41 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
|
||||||
|
|
||||||
# get results
|
# get results
|
||||||
def result(customer):
|
def result(customer):
|
||||||
return {'value': str(customer.card_number),
|
return {"value": str(customer.card_number), "label": str(customer)}
|
||||||
'label': str(customer)}
|
|
||||||
results = [result(c) for c in customers]
|
results = [result(c) for c in customers]
|
||||||
|
|
||||||
op_session.close()
|
op_session.close()
|
||||||
return results
|
return results
|
||||||
|
|
||||||
def refresh_batch_from_external_customer(self, batch):
|
def refresh_batch_from_external_customer( # pylint: disable=empty-docstring
|
||||||
|
self, batch
|
||||||
|
):
|
||||||
""" """
|
""" """
|
||||||
corepos = self.app.get_corepos_handler()
|
corepos = self.app.get_corepos_handler()
|
||||||
op_model = corepos.get_model_office_op()
|
op_model = corepos.get_model_office_op()
|
||||||
op_session = corepos.make_session_office_op()
|
op_session = corepos.make_session_office_op()
|
||||||
|
|
||||||
if not batch.customer_id.isdigit():
|
if not batch.customer_id.isdigit():
|
||||||
raise ValueError(f"invalid CORE-POS customer card number: {batch.customer_id}")
|
raise ValueError(
|
||||||
|
f"invalid CORE-POS customer card number: {batch.customer_id}"
|
||||||
|
)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
customer = op_session.query(op_model.CustomerClassic)\
|
customer = (
|
||||||
.join(op_model.MemberInfo,
|
op_session.query(op_model.CustomerClassic)
|
||||||
op_model.MemberInfo.card_number == op_model.CustomerClassic.card_number)\
|
.join(
|
||||||
.filter(op_model.CustomerClassic.card_number == int(batch.customer_id))\
|
op_model.MemberInfo,
|
||||||
.filter(op_model.CustomerClassic.person_number == 1)\
|
op_model.MemberInfo.card_number
|
||||||
.options(orm.joinedload(op_model.CustomerClassic.member_info))\
|
== op_model.CustomerClassic.card_number,
|
||||||
.one()
|
)
|
||||||
except orm.exc.NoResultFound:
|
.filter(op_model.CustomerClassic.card_number == int(batch.customer_id))
|
||||||
raise ValueError(f"CORE-POS Customer not found: {batch.customer_id}")
|
.filter(op_model.CustomerClassic.person_number == 1)
|
||||||
|
.options(orm.joinedload(op_model.CustomerClassic.member_info))
|
||||||
|
.one()
|
||||||
|
)
|
||||||
|
except orm.exc.NoResultFound as e:
|
||||||
|
raise ValueError(f"CORE-POS Customer not found: {batch.customer_id}") from e
|
||||||
|
|
||||||
batch.customer_name = str(customer)
|
batch.customer_name = str(customer)
|
||||||
batch.phone_number = customer.member_info.phone
|
batch.phone_number = customer.member_info.phone
|
||||||
|
@ -105,7 +121,9 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
|
||||||
|
|
||||||
op_session.close()
|
op_session.close()
|
||||||
|
|
||||||
def autocomplete_products_external(self, session, term, user=None):
|
def autocomplete_products_external( # pylint: disable=empty-docstring
|
||||||
|
self, session, term, user=None
|
||||||
|
):
|
||||||
""" """
|
""" """
|
||||||
corepos = self.app.get_corepos_handler()
|
corepos = self.app.get_corepos_handler()
|
||||||
op_model = corepos.get_model_office_op()
|
op_model = corepos.get_model_office_op()
|
||||||
|
@ -117,14 +135,16 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
|
||||||
# filter query
|
# filter query
|
||||||
criteria = []
|
criteria = []
|
||||||
for word in term.split():
|
for word in term.split():
|
||||||
criteria.append(sa.or_(
|
criteria.append(
|
||||||
op_model.Product.brand.ilike(f'%{word}%'),
|
sa.or_(
|
||||||
op_model.Product.description.ilike(f'%{word}%')))
|
op_model.Product.brand.ilike(f"%{word}%"),
|
||||||
|
op_model.Product.description.ilike(f"%{word}%"),
|
||||||
|
)
|
||||||
|
)
|
||||||
query = query.filter(sa.and_(*criteria))
|
query = query.filter(sa.and_(*criteria))
|
||||||
|
|
||||||
# sort query
|
# sort query
|
||||||
query = query.order_by(op_model.Product.brand,
|
query = query.order_by(op_model.Product.brand, op_model.Product.description)
|
||||||
op_model.Product.description)
|
|
||||||
|
|
||||||
# get data
|
# get data
|
||||||
# TODO: need max_results option
|
# TODO: need max_results option
|
||||||
|
@ -132,43 +152,50 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
|
||||||
|
|
||||||
# get results
|
# get results
|
||||||
def result(product):
|
def result(product):
|
||||||
return {'value': product.upc,
|
return {
|
||||||
'label': self.app.make_full_name(product.brand,
|
"value": product.upc,
|
||||||
product.description,
|
"label": self.app.make_full_name(
|
||||||
product.size)}
|
product.brand, product.description, product.size
|
||||||
|
),
|
||||||
|
}
|
||||||
|
|
||||||
results = [result(c) for c in products]
|
results = [result(c) for c in products]
|
||||||
|
|
||||||
op_session.close()
|
op_session.close()
|
||||||
return results
|
return results
|
||||||
|
|
||||||
def get_product_info_external(self, session, product_id, user=None):
|
def get_product_info_external( # pylint: disable=empty-docstring
|
||||||
|
self, session, product_id, user=None
|
||||||
|
):
|
||||||
""" """
|
""" """
|
||||||
corepos = self.app.get_corepos_handler()
|
corepos = self.app.get_corepos_handler()
|
||||||
op_model = corepos.get_model_office_op()
|
op_model = corepos.get_model_office_op()
|
||||||
op_session = corepos.make_session_office_op()
|
op_session = corepos.make_session_office_op()
|
||||||
|
|
||||||
try:
|
try:
|
||||||
product = op_session.query(op_model.Product)\
|
product = (
|
||||||
.filter(op_model.Product.upc == product_id)\
|
op_session.query(op_model.Product)
|
||||||
.one()
|
.filter(op_model.Product.upc == product_id)
|
||||||
except orm.exc.NoResultFound:
|
.one()
|
||||||
raise ValueError(f"CORE-POS Product not found: {product_id}")
|
)
|
||||||
|
except orm.exc.NoResultFound as e:
|
||||||
|
raise ValueError(f"CORE-POS Product not found: {product_id}") from e
|
||||||
|
|
||||||
data = {
|
data = {
|
||||||
'product_id': product.upc,
|
"product_id": product.upc,
|
||||||
'scancode': product.upc,
|
"scancode": product.upc,
|
||||||
'brand_name': product.brand,
|
"brand_name": product.brand,
|
||||||
'description': product.description,
|
"description": product.description,
|
||||||
'size': product.size,
|
"size": product.size,
|
||||||
'full_description': self.app.make_full_name(product.brand,
|
"full_description": self.app.make_full_name(
|
||||||
product.description,
|
product.brand, product.description, product.size
|
||||||
product.size),
|
),
|
||||||
'weighed': product.scale,
|
"weighed": product.scale,
|
||||||
'special_order': False,
|
"special_order": False,
|
||||||
'department_id': product.department_number,
|
"department_id": product.department_number,
|
||||||
'department_name': product.department.name if product.department else None,
|
"department_name": product.department.name if product.department else None,
|
||||||
'case_size': self.get_case_size_for_external_product(product),
|
"case_size": self.get_case_size_for_external_product(product),
|
||||||
'unit_price_reg': self.get_unit_price_reg_for_external_product(product),
|
"unit_price_reg": self.get_unit_price_reg_for_external_product(product),
|
||||||
# TODO
|
# TODO
|
||||||
# 'vendor_name': product.vendor_name,
|
# 'vendor_name': product.vendor_name,
|
||||||
# 'vendor_item_code': product.vendor_item_code,
|
# 'vendor_item_code': product.vendor_item_code,
|
||||||
|
@ -177,18 +204,20 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
|
||||||
op_session.close()
|
op_session.close()
|
||||||
return data
|
return data
|
||||||
|
|
||||||
def refresh_row_from_external_product(self, row):
|
def refresh_row_from_external_product(self, row): # pylint: disable=empty-docstring
|
||||||
""" """
|
""" """
|
||||||
corepos = self.app.get_corepos_handler()
|
corepos = self.app.get_corepos_handler()
|
||||||
op_model = corepos.get_model_office_op()
|
op_model = corepos.get_model_office_op()
|
||||||
op_session = corepos.make_session_office_op()
|
op_session = corepos.make_session_office_op()
|
||||||
|
|
||||||
try:
|
try:
|
||||||
product = op_session.query(op_model.Product)\
|
product = (
|
||||||
.filter(op_model.Product.upc == row.product_id)\
|
op_session.query(op_model.Product)
|
||||||
.one()
|
.filter(op_model.Product.upc == row.product_id)
|
||||||
except orm.exc.NoResultFound:
|
.one()
|
||||||
raise ValueError(f"CORE-POS Product not found: {row.product_id}")
|
)
|
||||||
|
except orm.exc.NoResultFound as e:
|
||||||
|
raise ValueError(f"CORE-POS Product not found: {row.product_id}") from e
|
||||||
|
|
||||||
row.product_scancode = product.upc
|
row.product_scancode = product.upc
|
||||||
row.product_brand = product.brand
|
row.product_brand = product.brand
|
||||||
|
@ -198,20 +227,34 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
|
||||||
row.department_id = product.department_number
|
row.department_id = product.department_number
|
||||||
row.department_name = product.department.name if product.department else None
|
row.department_name = product.department.name if product.department else None
|
||||||
row.special_order = False
|
row.special_order = False
|
||||||
|
|
||||||
|
row.vendor_name = None
|
||||||
|
row.vendor_item_code = None
|
||||||
|
item = product.default_vendor_item
|
||||||
|
if item:
|
||||||
|
row.vendor_name = item.vendor.name if item.vendor else None
|
||||||
|
row.vendor_item_code = item.sku
|
||||||
|
|
||||||
row.case_size = self.get_case_size_for_external_product(product)
|
row.case_size = self.get_case_size_for_external_product(product)
|
||||||
row.unit_cost = product.cost
|
row.unit_cost = product.cost
|
||||||
row.unit_price_reg = self.get_unit_price_reg_for_external_product(product)
|
row.unit_price_reg = self.get_unit_price_reg_for_external_product(product)
|
||||||
|
|
||||||
op_session.close()
|
op_session.close()
|
||||||
|
|
||||||
def get_case_size_for_external_product(self, product):
|
def get_case_size_for_external_product( # pylint: disable=empty-docstring
|
||||||
|
self, product
|
||||||
|
):
|
||||||
""" """
|
""" """
|
||||||
if product.vendor_items:
|
if product.vendor_items:
|
||||||
item = product.vendor_items[0]
|
item = product.vendor_items[0]
|
||||||
if item.units is not None:
|
if item.units is not None:
|
||||||
return decimal.Decimal(f'{item.units:0.4f}')
|
return decimal.Decimal(f"{item.units:0.4f}")
|
||||||
|
return None
|
||||||
|
|
||||||
def get_unit_price_reg_for_external_product(self, product):
|
def get_unit_price_reg_for_external_product( # pylint: disable=empty-docstring
|
||||||
|
self, product
|
||||||
|
):
|
||||||
""" """
|
""" """
|
||||||
if product.normal_price is not None:
|
if product.normal_price is not None:
|
||||||
return decimal.Decimal(f'{product.normal_price:0.3f}')
|
return decimal.Decimal(f"{product.normal_price:0.3f}")
|
||||||
|
return None
|
||||||
|
|
|
@ -33,15 +33,20 @@ class SideshowCoreposConfig(WuttaConfigExtension):
|
||||||
|
|
||||||
This establishes some config defaults specific to Sideshow-COREPOS.
|
This establishes some config defaults specific to Sideshow-COREPOS.
|
||||||
"""
|
"""
|
||||||
key = 'sideshow_corepos'
|
|
||||||
|
|
||||||
def configure(self, config):
|
key = "sideshow_corepos"
|
||||||
|
|
||||||
|
def configure(self, config): # pylint: disable=empty-docstring
|
||||||
""" """
|
""" """
|
||||||
|
|
||||||
# batch handlers
|
# batch handlers
|
||||||
config.setdefault(f'{config.appname}.batch.neworder.handler.spec',
|
config.setdefault(
|
||||||
'sideshow_corepos.batch.neworder:NewOrderBatchHandler')
|
f"{config.appname}.batch.neworder.handler.spec",
|
||||||
|
"sideshow_corepos.batch.neworder:NewOrderBatchHandler",
|
||||||
|
)
|
||||||
|
|
||||||
# web app menu
|
# web app menu
|
||||||
config.setdefault(f'{config.appname}.web.menus.handler.spec',
|
config.setdefault(
|
||||||
'sideshow_corepos.web.menus:SideshowMenuHandler')
|
f"{config.appname}.web.menus.handler.spec",
|
||||||
|
"sideshow_corepos.web.menus:SideshowMenuHandler",
|
||||||
|
)
|
||||||
|
|
|
@ -25,5 +25,5 @@ Sideshow-COREPOS - Case/Special Order Tracker for CORE-POS
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
|
||||||
def includeme(config):
|
def includeme(config): # pylint: disable=missing-function-docstring
|
||||||
config.include('sideshow_corepos.web.views')
|
config.include("sideshow_corepos.web.views")
|
||||||
|
|
|
@ -29,39 +29,43 @@ from wuttaweb import app as base
|
||||||
from wutta_corepos.web.db import CoreOpSession
|
from wutta_corepos.web.db import CoreOpSession
|
||||||
|
|
||||||
|
|
||||||
def main(global_config, **settings):
|
def main(global_config, **settings): # pylint: disable=unused-argument
|
||||||
"""
|
"""
|
||||||
Make and return the WSGI app (Paste entry point).
|
Make and return the WSGI app (Paste entry point).
|
||||||
"""
|
"""
|
||||||
# prefer Sideshow templates over wuttaweb
|
# prefer Sideshow templates over wuttaweb
|
||||||
settings.setdefault('mako.directories', [
|
settings.setdefault(
|
||||||
'sideshow.web:templates',
|
"mako.directories",
|
||||||
'wuttaweb:templates',
|
[
|
||||||
])
|
"sideshow.web:templates",
|
||||||
|
"wuttaweb:templates",
|
||||||
|
],
|
||||||
|
)
|
||||||
|
|
||||||
# make config objects
|
# make config objects
|
||||||
wutta_config = base.make_wutta_config(settings)
|
wutta_config = base.make_wutta_config(settings)
|
||||||
pyramid_config = base.make_pyramid_config(settings)
|
pyramid_config = base.make_pyramid_config(settings)
|
||||||
|
|
||||||
# configure DB sessions
|
# configure DB sessions
|
||||||
CoreOpSession.configure(bind=wutta_config.core_office_op_engine)
|
if hasattr(wutta_config, "core_office_op_engine"):
|
||||||
|
CoreOpSession.configure(bind=wutta_config.core_office_op_engine)
|
||||||
|
|
||||||
# bring in the rest of Sideshow
|
# bring in the rest of Sideshow
|
||||||
pyramid_config.include('sideshow.web')
|
pyramid_config.include("sideshow.web")
|
||||||
pyramid_config.include('sideshow_corepos.web')
|
pyramid_config.include("sideshow_corepos.web")
|
||||||
|
|
||||||
return pyramid_config.make_wsgi_app()
|
return pyramid_config.make_wsgi_app()
|
||||||
|
|
||||||
|
|
||||||
def make_wsgi_app():
|
def make_wsgi_app(config=None):
|
||||||
"""
|
"""
|
||||||
Make and return the WSGI app (generic entry point).
|
Make and return the WSGI app (generic entry point).
|
||||||
"""
|
"""
|
||||||
return base.make_wsgi_app(main)
|
return base.make_wsgi_app(main, config=config)
|
||||||
|
|
||||||
|
|
||||||
def make_asgi_app():
|
def make_asgi_app(config=None):
|
||||||
"""
|
"""
|
||||||
Make and return the ASGI app (generic entry point).
|
Make and return the ASGI app (generic entry point).
|
||||||
"""
|
"""
|
||||||
return base.make_asgi_app(main)
|
return base.make_asgi_app(main, config=config)
|
||||||
|
|
|
@ -38,14 +38,16 @@ class SideshowMenuHandler(base.SideshowMenuHandler):
|
||||||
"""
|
"""
|
||||||
menu = super().make_customers_menu(request, **kwargs)
|
menu = super().make_customers_menu(request, **kwargs)
|
||||||
|
|
||||||
menu['items'].extend([
|
menu["items"].extend(
|
||||||
{'type': 'sep'},
|
[
|
||||||
{
|
{"type": "sep"},
|
||||||
'title': "CORE-POS Members",
|
{
|
||||||
'route': 'corepos_members',
|
"title": "CORE-POS Members",
|
||||||
'perm': 'corepos_members.list',
|
"route": "corepos_members",
|
||||||
},
|
"perm": "corepos_members.list",
|
||||||
])
|
},
|
||||||
|
]
|
||||||
|
)
|
||||||
|
|
||||||
return menu
|
return menu
|
||||||
|
|
||||||
|
@ -55,14 +57,16 @@ class SideshowMenuHandler(base.SideshowMenuHandler):
|
||||||
"""
|
"""
|
||||||
menu = super().make_products_menu(request, **kwargs)
|
menu = super().make_products_menu(request, **kwargs)
|
||||||
|
|
||||||
menu['items'].extend([
|
menu["items"].extend(
|
||||||
{'type': 'sep'},
|
[
|
||||||
{
|
{"type": "sep"},
|
||||||
'title': "CORE-POS Products",
|
{
|
||||||
'route': 'corepos_products',
|
"title": "CORE-POS Products",
|
||||||
'perm': 'corepos_products.list',
|
"route": "corepos_products",
|
||||||
},
|
"perm": "corepos_products.list",
|
||||||
])
|
},
|
||||||
|
]
|
||||||
|
)
|
||||||
|
|
||||||
return menu
|
return menu
|
||||||
|
|
||||||
|
@ -75,12 +79,14 @@ class SideshowMenuHandler(base.SideshowMenuHandler):
|
||||||
corepos = self.app.get_corepos_handler()
|
corepos = self.app.get_corepos_handler()
|
||||||
url = corepos.get_office_url()
|
url = corepos.get_office_url()
|
||||||
if url:
|
if url:
|
||||||
menu['items'].extend([
|
menu["items"].extend(
|
||||||
{
|
[
|
||||||
'title': "CORE Office",
|
{
|
||||||
'url': url,
|
"title": "CORE Office",
|
||||||
'target': '_blank',
|
"url": url,
|
||||||
},
|
"target": "_blank",
|
||||||
])
|
},
|
||||||
|
]
|
||||||
|
)
|
||||||
|
|
||||||
return menu
|
return menu
|
||||||
|
|
|
@ -27,8 +27,8 @@ This adds config for readonly views for CORE-POS members and products.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
|
||||||
def includeme(config):
|
def includeme(config): # pylint: disable=missing-function-docstring
|
||||||
|
|
||||||
# CORE-POS views
|
# CORE-POS views
|
||||||
config.include('wutta_corepos.web.views.corepos.members')
|
config.include("wutta_corepos.web.views.corepos.members")
|
||||||
config.include('wutta_corepos.web.views.corepos.products')
|
config.include("wutta_corepos.web.views.corepos.products")
|
||||||
|
|
14
tasks.py
14
tasks.py
|
@ -15,14 +15,14 @@ def release(c, skip_tests=False):
|
||||||
Release a new version of Sideshow-COREPOS
|
Release a new version of Sideshow-COREPOS
|
||||||
"""
|
"""
|
||||||
if not skip_tests:
|
if not skip_tests:
|
||||||
c.run('pytest')
|
c.run("pytest")
|
||||||
|
|
||||||
# rebuild pkg
|
# rebuild pkg
|
||||||
if os.path.exists('dist'):
|
if os.path.exists("dist"):
|
||||||
shutil.rmtree('dist')
|
shutil.rmtree("dist")
|
||||||
if os.path.exists('Sideshow_COREPOS.egg-info'):
|
if os.path.exists("Sideshow_COREPOS.egg-info"):
|
||||||
shutil.rmtree('Sideshow_COREPOS.egg-info')
|
shutil.rmtree("Sideshow_COREPOS.egg-info")
|
||||||
c.run('python -m build --sdist')
|
c.run("python -m build --sdist")
|
||||||
|
|
||||||
# upload
|
# upload
|
||||||
c.run('twine upload dist/*')
|
c.run("twine upload dist/*")
|
||||||
|
|
|
@ -17,8 +17,8 @@ class TestNewOrderBatchHandler(DataTestCase):
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
super().setUp()
|
super().setUp()
|
||||||
|
|
||||||
self.op_engine = sa.create_engine('sqlite://')
|
self.op_engine = sa.create_engine("sqlite://")
|
||||||
self.config.core_office_op_engines = {'default': self.op_engine}
|
self.config.core_office_op_engines = {"default": self.op_engine}
|
||||||
self.config.core_office_op_engine = self.op_engine
|
self.config.core_office_op_engine = self.op_engine
|
||||||
|
|
||||||
op_model.Base.metadata.create_all(bind=self.op_engine)
|
op_model.Base.metadata.create_all(bind=self.op_engine)
|
||||||
|
@ -31,7 +31,7 @@ class TestNewOrderBatchHandler(DataTestCase):
|
||||||
|
|
||||||
def make_config(self, **kwargs):
|
def make_config(self, **kwargs):
|
||||||
config = super().make_config(**kwargs)
|
config = super().make_config(**kwargs)
|
||||||
config.setdefault('wutta.enum_spec', 'sideshow.enum')
|
config.setdefault("wutta.enum_spec", "sideshow.enum")
|
||||||
return config
|
return config
|
||||||
|
|
||||||
def make_handler(self):
|
def make_handler(self):
|
||||||
|
@ -41,33 +41,41 @@ class TestNewOrderBatchHandler(DataTestCase):
|
||||||
handler = self.make_handler()
|
handler = self.make_handler()
|
||||||
|
|
||||||
# empty results by default
|
# empty results by default
|
||||||
self.assertEqual(handler.autocomplete_customers_external(self.session, 'foo'), [])
|
self.assertEqual(
|
||||||
|
handler.autocomplete_customers_external(self.session, "foo"), []
|
||||||
|
)
|
||||||
|
|
||||||
# add a member
|
# add a member
|
||||||
member = op_model.MemberInfo(card_number=42)
|
member = op_model.MemberInfo(card_number=42)
|
||||||
self.op_session.add(member)
|
self.op_session.add(member)
|
||||||
customer = op_model.CustomerClassic(first_name="Chuck", last_name="Norris",
|
customer = op_model.CustomerClassic(
|
||||||
last_change=datetime.datetime.now())
|
first_name="Chuck", last_name="Norris", last_change=datetime.datetime.now()
|
||||||
|
)
|
||||||
member.customers.append(customer)
|
member.customers.append(customer)
|
||||||
self.op_session.add(customer)
|
self.op_session.add(customer)
|
||||||
self.op_session.flush()
|
self.op_session.flush()
|
||||||
|
|
||||||
# search for chuck finds chuck
|
# search for chuck finds chuck
|
||||||
results = handler.autocomplete_customers_external(self.session, 'chuck')
|
results = handler.autocomplete_customers_external(self.session, "chuck")
|
||||||
self.assertEqual(len(results), 1)
|
self.assertEqual(len(results), 1)
|
||||||
self.assertEqual(results[0], {
|
self.assertEqual(
|
||||||
'value': '42',
|
results[0],
|
||||||
'label': "Chuck Norris",
|
{
|
||||||
})
|
"value": "42",
|
||||||
|
"label": "Chuck Norris",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
# search for sally finds nothing
|
# search for sally finds nothing
|
||||||
self.assertEqual(handler.autocomplete_customers_external(self.session, 'sally'), [])
|
self.assertEqual(
|
||||||
|
handler.autocomplete_customers_external(self.session, "sally"), []
|
||||||
|
)
|
||||||
|
|
||||||
def test_refresh_batch_from_external_customer(self):
|
def test_refresh_batch_from_external_customer(self):
|
||||||
model = self.app.model
|
model = self.app.model
|
||||||
handler = self.make_handler()
|
handler = self.make_handler()
|
||||||
|
|
||||||
user = model.User(username='barney')
|
user = model.User(username="barney")
|
||||||
self.session.add(user)
|
self.session.add(user)
|
||||||
self.session.flush()
|
self.session.flush()
|
||||||
|
|
||||||
|
@ -76,91 +84,114 @@ class TestNewOrderBatchHandler(DataTestCase):
|
||||||
self.session.flush()
|
self.session.flush()
|
||||||
|
|
||||||
# add a member
|
# add a member
|
||||||
member = op_model.MemberInfo(card_number=42, phone='555-1234', email='chuck@example.com')
|
member = op_model.MemberInfo(
|
||||||
|
card_number=42, phone="555-1234", email="chuck@example.com"
|
||||||
|
)
|
||||||
self.op_session.add(member)
|
self.op_session.add(member)
|
||||||
customer = op_model.CustomerClassic(first_name="Chuck", last_name="Norris",
|
customer = op_model.CustomerClassic(
|
||||||
last_change=datetime.datetime.now())
|
first_name="Chuck", last_name="Norris", last_change=datetime.datetime.now()
|
||||||
|
)
|
||||||
member.customers.append(customer)
|
member.customers.append(customer)
|
||||||
self.op_session.add(customer)
|
self.op_session.add(customer)
|
||||||
self.op_session.flush()
|
self.op_session.flush()
|
||||||
|
|
||||||
# error if invalid customer_id
|
# error if invalid customer_id
|
||||||
batch.customer_id = 'BreakThings!'
|
batch.customer_id = "BreakThings!"
|
||||||
self.assertRaises(ValueError, handler.refresh_batch_from_external_customer, batch)
|
self.assertRaises(
|
||||||
|
ValueError, handler.refresh_batch_from_external_customer, batch
|
||||||
|
)
|
||||||
|
|
||||||
# error if customer not found
|
# error if customer not found
|
||||||
batch.customer_id = '9999'
|
batch.customer_id = "9999"
|
||||||
self.assertRaises(ValueError, handler.refresh_batch_from_external_customer, batch)
|
self.assertRaises(
|
||||||
|
ValueError, handler.refresh_batch_from_external_customer, batch
|
||||||
|
)
|
||||||
|
|
||||||
# batch should reflect customer info
|
# batch should reflect customer info
|
||||||
batch.customer_id = '42'
|
batch.customer_id = "42"
|
||||||
self.assertIsNone(batch.customer_name)
|
self.assertIsNone(batch.customer_name)
|
||||||
self.assertIsNone(batch.phone_number)
|
self.assertIsNone(batch.phone_number)
|
||||||
self.assertIsNone(batch.email_address)
|
self.assertIsNone(batch.email_address)
|
||||||
handler.refresh_batch_from_external_customer(batch)
|
handler.refresh_batch_from_external_customer(batch)
|
||||||
self.assertEqual(batch.customer_name, "Chuck Norris")
|
self.assertEqual(batch.customer_name, "Chuck Norris")
|
||||||
self.assertEqual(batch.phone_number, '555-1234')
|
self.assertEqual(batch.phone_number, "555-1234")
|
||||||
self.assertEqual(batch.email_address, 'chuck@example.com')
|
self.assertEqual(batch.email_address, "chuck@example.com")
|
||||||
|
|
||||||
def test_autocomplete_products_local(self):
|
def test_autocomplete_products_local(self):
|
||||||
handler = self.make_handler()
|
handler = self.make_handler()
|
||||||
|
|
||||||
# empty results by default
|
# empty results by default
|
||||||
self.assertEqual(handler.autocomplete_products_external(self.session, 'foo'), [])
|
self.assertEqual(
|
||||||
|
handler.autocomplete_products_external(self.session, "foo"), []
|
||||||
|
)
|
||||||
|
|
||||||
# add a product
|
# add a product
|
||||||
product = op_model.Product(upc='07430500132', brand="Bragg's",
|
product = op_model.Product(
|
||||||
description="Vinegar", size='32oz')
|
upc="07430500132", brand="Bragg's", description="Vinegar", size="32oz"
|
||||||
|
)
|
||||||
self.op_session.add(product)
|
self.op_session.add(product)
|
||||||
self.op_session.commit()
|
self.op_session.commit()
|
||||||
|
|
||||||
# search for vinegar finds product
|
# search for vinegar finds product
|
||||||
results = handler.autocomplete_products_external(self.session, 'vinegar')
|
results = handler.autocomplete_products_external(self.session, "vinegar")
|
||||||
self.assertEqual(len(results), 1)
|
self.assertEqual(len(results), 1)
|
||||||
self.assertEqual(results[0], {
|
self.assertEqual(
|
||||||
'value': '07430500132',
|
results[0],
|
||||||
'label': "Bragg's Vinegar 32oz",
|
{
|
||||||
})
|
"value": "07430500132",
|
||||||
|
"label": "Bragg's Vinegar 32oz",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
# search for brag finds product
|
# search for brag finds product
|
||||||
results = handler.autocomplete_products_external(self.session, 'brag')
|
results = handler.autocomplete_products_external(self.session, "brag")
|
||||||
self.assertEqual(len(results), 1)
|
self.assertEqual(len(results), 1)
|
||||||
self.assertEqual(results[0], {
|
self.assertEqual(
|
||||||
'value': '07430500132',
|
results[0],
|
||||||
'label': "Bragg's Vinegar 32oz",
|
{
|
||||||
})
|
"value": "07430500132",
|
||||||
|
"label": "Bragg's Vinegar 32oz",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
# search for juice finds nothing
|
# search for juice finds nothing
|
||||||
self.assertEqual(handler.autocomplete_products_external(self.session, 'juice'), [])
|
self.assertEqual(
|
||||||
|
handler.autocomplete_products_external(self.session, "juice"), []
|
||||||
|
)
|
||||||
|
|
||||||
def test_get_case_size_for_external_product(self):
|
def test_get_case_size_for_external_product(self):
|
||||||
handler = self.make_handler()
|
handler = self.make_handler()
|
||||||
|
|
||||||
# null
|
# null
|
||||||
product = op_model.Product(upc='07430500132', brand="Bragg's",
|
product = op_model.Product(
|
||||||
description="Vinegar", size='32oz')
|
upc="07430500132", brand="Bragg's", description="Vinegar", size="32oz"
|
||||||
|
)
|
||||||
self.op_session.add(product)
|
self.op_session.add(product)
|
||||||
self.op_session.commit()
|
self.op_session.commit()
|
||||||
self.op_session.refresh(product)
|
self.op_session.refresh(product)
|
||||||
self.assertIsNone(handler.get_case_size_for_external_product(product))
|
self.assertIsNone(handler.get_case_size_for_external_product(product))
|
||||||
|
|
||||||
# typical
|
# typical
|
||||||
vendor = op_model.Vendor(id=42, name='Acme Distributors')
|
vendor = op_model.Vendor(id=42, name="Acme Distributors")
|
||||||
self.op_session.add(vendor)
|
self.op_session.add(vendor)
|
||||||
item = op_model.VendorItem(vendor=vendor, sku='1234', units=12.34,
|
item = op_model.VendorItem(
|
||||||
vendor_item_id=1)
|
vendor=vendor, sku="1234", units=12.34, vendor_item_id=1
|
||||||
|
)
|
||||||
product.vendor_items.append(item)
|
product.vendor_items.append(item)
|
||||||
self.op_session.commit()
|
self.op_session.commit()
|
||||||
self.op_session.refresh(product)
|
self.op_session.refresh(product)
|
||||||
self.assertEqual(handler.get_case_size_for_external_product(product),
|
self.assertEqual(
|
||||||
decimal.Decimal('12.3400'))
|
handler.get_case_size_for_external_product(product),
|
||||||
|
decimal.Decimal("12.3400"),
|
||||||
|
)
|
||||||
|
|
||||||
def test_get_unit_price_reg_for_external_product(self):
|
def test_get_unit_price_reg_for_external_product(self):
|
||||||
handler = self.make_handler()
|
handler = self.make_handler()
|
||||||
|
|
||||||
# null
|
# null
|
||||||
product = op_model.Product(upc='07430500132', brand="Bragg's",
|
product = op_model.Product(
|
||||||
description="Vinegar", size='32oz')
|
upc="07430500132", brand="Bragg's", description="Vinegar", size="32oz"
|
||||||
|
)
|
||||||
self.op_session.add(product)
|
self.op_session.add(product)
|
||||||
self.op_session.commit()
|
self.op_session.commit()
|
||||||
self.op_session.refresh(product)
|
self.op_session.refresh(product)
|
||||||
|
@ -170,53 +201,64 @@ class TestNewOrderBatchHandler(DataTestCase):
|
||||||
product.normal_price = 4.19
|
product.normal_price = 4.19
|
||||||
self.op_session.commit()
|
self.op_session.commit()
|
||||||
self.op_session.refresh(product)
|
self.op_session.refresh(product)
|
||||||
self.assertEqual(handler.get_unit_price_reg_for_external_product(product),
|
self.assertEqual(
|
||||||
decimal.Decimal('4.19'))
|
handler.get_unit_price_reg_for_external_product(product),
|
||||||
|
decimal.Decimal("4.19"),
|
||||||
|
)
|
||||||
|
|
||||||
def test_get_product_info_external(self):
|
def test_get_product_info_external(self):
|
||||||
model = self.app.model
|
model = self.app.model
|
||||||
handler = self.make_handler()
|
handler = self.make_handler()
|
||||||
|
|
||||||
user = model.User(username='barney')
|
user = model.User(username="barney")
|
||||||
self.session.add(user)
|
self.session.add(user)
|
||||||
batch = handler.make_batch(self.session, created_by=user)
|
batch = handler.make_batch(self.session, created_by=user)
|
||||||
self.session.add(batch)
|
self.session.add(batch)
|
||||||
self.session.flush()
|
self.session.flush()
|
||||||
|
|
||||||
vendor = op_model.Vendor(id=42, name='Acme Distributors')
|
vendor = op_model.Vendor(id=42, name="Acme Distributors")
|
||||||
self.op_session.add(vendor)
|
self.op_session.add(vendor)
|
||||||
product = op_model.Product(upc='07430500132', brand="Bragg",
|
product = op_model.Product(
|
||||||
description="Vinegar", size='32oz',
|
upc="07430500132",
|
||||||
normal_price=4.19)
|
brand="Bragg",
|
||||||
item = op_model.VendorItem(vendor=vendor, sku='1234', units=12.34,
|
description="Vinegar",
|
||||||
vendor_item_id=1)
|
size="32oz",
|
||||||
|
normal_price=4.19,
|
||||||
|
)
|
||||||
|
item = op_model.VendorItem(
|
||||||
|
vendor=vendor, sku="1234", units=12.34, vendor_item_id=1
|
||||||
|
)
|
||||||
product.vendor_items.append(item)
|
product.vendor_items.append(item)
|
||||||
self.op_session.add(product)
|
self.op_session.add(product)
|
||||||
self.op_session.commit()
|
self.op_session.commit()
|
||||||
|
|
||||||
# typical
|
# typical
|
||||||
info = handler.get_product_info_external(self.session, '07430500132')
|
info = handler.get_product_info_external(self.session, "07430500132")
|
||||||
self.assertEqual(info['product_id'], '07430500132')
|
self.assertEqual(info["product_id"], "07430500132")
|
||||||
self.assertEqual(info['scancode'], '07430500132')
|
self.assertEqual(info["scancode"], "07430500132")
|
||||||
self.assertEqual(info['brand_name'], 'Bragg')
|
self.assertEqual(info["brand_name"], "Bragg")
|
||||||
self.assertEqual(info['description'], 'Vinegar')
|
self.assertEqual(info["description"], "Vinegar")
|
||||||
self.assertEqual(info['size'], '32oz')
|
self.assertEqual(info["size"], "32oz")
|
||||||
self.assertEqual(info['full_description'], 'Bragg Vinegar 32oz')
|
self.assertEqual(info["full_description"], "Bragg Vinegar 32oz")
|
||||||
self.assertEqual(info['case_size'], decimal.Decimal('12.3400'))
|
self.assertEqual(info["case_size"], decimal.Decimal("12.3400"))
|
||||||
self.assertEqual(info['unit_price_reg'], decimal.Decimal('4.19'))
|
self.assertEqual(info["unit_price_reg"], decimal.Decimal("4.19"))
|
||||||
|
|
||||||
# error if no product_id
|
# error if no product_id
|
||||||
self.assertRaises(ValueError, handler.get_product_info_external, self.session, None)
|
self.assertRaises(
|
||||||
|
ValueError, handler.get_product_info_external, self.session, None
|
||||||
|
)
|
||||||
|
|
||||||
# error if product not found
|
# error if product not found
|
||||||
self.assertRaises(ValueError, handler.get_product_info_external, self.session, 'BADUPC')
|
self.assertRaises(
|
||||||
|
ValueError, handler.get_product_info_external, self.session, "BADUPC"
|
||||||
|
)
|
||||||
|
|
||||||
def test_refresh_row_from_external_product(self):
|
def test_refresh_row_from_external_product(self):
|
||||||
model = self.app.model
|
model = self.app.model
|
||||||
enum = self.app.enum
|
enum = self.app.enum
|
||||||
handler = self.make_handler()
|
handler = self.make_handler()
|
||||||
|
|
||||||
user = model.User(username='barney')
|
user = model.User(username="barney")
|
||||||
self.session.add(user)
|
self.session.add(user)
|
||||||
batch = handler.make_batch(self.session, created_by=user)
|
batch = handler.make_batch(self.session, created_by=user)
|
||||||
self.session.add(batch)
|
self.session.add(batch)
|
||||||
|
@ -225,27 +267,32 @@ class TestNewOrderBatchHandler(DataTestCase):
|
||||||
self.session.add(row)
|
self.session.add(row)
|
||||||
self.session.flush()
|
self.session.flush()
|
||||||
|
|
||||||
vendor = op_model.Vendor(id=42, name='Acme Distributors')
|
vendor = op_model.Vendor(id=42, name="Acme Distributors")
|
||||||
self.op_session.add(vendor)
|
self.op_session.add(vendor)
|
||||||
product = op_model.Product(upc='07430500132', brand="Bragg",
|
product = op_model.Product(
|
||||||
description="Vinegar", size='32oz',
|
upc="07430500132",
|
||||||
normal_price=4.19)
|
brand="Bragg",
|
||||||
item = op_model.VendorItem(vendor=vendor, sku='1234', units=12.34,
|
description="Vinegar",
|
||||||
vendor_item_id=1)
|
size="32oz",
|
||||||
|
normal_price=4.19,
|
||||||
|
)
|
||||||
|
item = op_model.VendorItem(
|
||||||
|
vendor=vendor, sku="1234", units=12.34, vendor_item_id=1
|
||||||
|
)
|
||||||
product.vendor_items.append(item)
|
product.vendor_items.append(item)
|
||||||
self.op_session.add(product)
|
self.op_session.add(product)
|
||||||
self.op_session.commit()
|
self.op_session.commit()
|
||||||
|
|
||||||
# error if invalid product_id
|
# error if invalid product_id
|
||||||
row.product_id = 'BreakThings!'
|
row.product_id = "BreakThings!"
|
||||||
self.assertRaises(ValueError, handler.refresh_row_from_external_product, row)
|
self.assertRaises(ValueError, handler.refresh_row_from_external_product, row)
|
||||||
|
|
||||||
# error if product not found
|
# error if product not found
|
||||||
row.product_id = '9999'
|
row.product_id = "9999"
|
||||||
self.assertRaises(ValueError, handler.refresh_row_from_external_product, row)
|
self.assertRaises(ValueError, handler.refresh_row_from_external_product, row)
|
||||||
|
|
||||||
# row should reflect product info
|
# row should reflect product info
|
||||||
row.product_id = '07430500132'
|
row.product_id = "07430500132"
|
||||||
self.assertIsNone(row.product_scancode)
|
self.assertIsNone(row.product_scancode)
|
||||||
self.assertIsNone(row.product_brand)
|
self.assertIsNone(row.product_brand)
|
||||||
self.assertIsNone(row.product_description)
|
self.assertIsNone(row.product_description)
|
||||||
|
@ -253,9 +300,11 @@ class TestNewOrderBatchHandler(DataTestCase):
|
||||||
self.assertIsNone(row.case_size)
|
self.assertIsNone(row.case_size)
|
||||||
self.assertIsNone(row.unit_price_reg)
|
self.assertIsNone(row.unit_price_reg)
|
||||||
handler.refresh_row_from_external_product(row)
|
handler.refresh_row_from_external_product(row)
|
||||||
self.assertEqual(row.product_scancode, '07430500132')
|
self.assertEqual(row.product_scancode, "07430500132")
|
||||||
self.assertEqual(row.product_brand, "Bragg")
|
self.assertEqual(row.product_brand, "Bragg")
|
||||||
self.assertEqual(row.product_description, "Vinegar")
|
self.assertEqual(row.product_description, "Vinegar")
|
||||||
self.assertEqual(row.product_size, "32oz")
|
self.assertEqual(row.product_size, "32oz")
|
||||||
self.assertEqual(row.case_size, decimal.Decimal('12.3400'))
|
self.assertEqual(row.case_size, decimal.Decimal("12.3400"))
|
||||||
self.assertEqual(row.unit_price_reg, decimal.Decimal('4.19'))
|
self.assertEqual(row.unit_price_reg, decimal.Decimal("4.19"))
|
||||||
|
self.assertEqual(row.vendor_name, "Acme Distributors")
|
||||||
|
self.assertEqual(row.vendor_item_code, "1234")
|
||||||
|
|
20
tests/test_config.py
Normal file
20
tests/test_config.py
Normal file
|
@ -0,0 +1,20 @@
|
||||||
|
# -*- coding: utf-8; -*-
|
||||||
|
|
||||||
|
from wuttjamaican.testing import ConfigTestCase
|
||||||
|
|
||||||
|
from sideshow_corepos import config as mod
|
||||||
|
|
||||||
|
|
||||||
|
class TestSideshowCoreposConfig(ConfigTestCase):
|
||||||
|
|
||||||
|
def test_basic(self):
|
||||||
|
|
||||||
|
self.assertIsNone(self.config.get("wutta.batch.neworder.handler.spec"))
|
||||||
|
|
||||||
|
ext = mod.SideshowCoreposConfig()
|
||||||
|
ext.configure(self.config)
|
||||||
|
|
||||||
|
self.assertEqual(
|
||||||
|
self.config.get("wutta.batch.neworder.handler.spec"),
|
||||||
|
"sideshow_corepos.batch.neworder:NewOrderBatchHandler",
|
||||||
|
)
|
|
@ -1,32 +1,44 @@
|
||||||
# -*- coding: utf-8; -*-
|
# -*- coding: utf-8; -*-
|
||||||
|
|
||||||
from wuttjamaican.testing import FileTestCase, ConfigTestCase
|
import sqlalchemy as sa
|
||||||
|
|
||||||
|
from wuttjamaican.testing import DataTestCase
|
||||||
|
|
||||||
from asgiref.wsgi import WsgiToAsgi
|
from asgiref.wsgi import WsgiToAsgi
|
||||||
from pyramid.router import Router
|
from pyramid.router import Router
|
||||||
|
|
||||||
|
from wutta_corepos.web.db import CoreOpSession
|
||||||
|
|
||||||
from sideshow_corepos.web import app as mod
|
from sideshow_corepos.web import app as mod
|
||||||
|
|
||||||
|
|
||||||
class TestMain(FileTestCase):
|
class TestMain(DataTestCase):
|
||||||
|
|
||||||
def test_basic(self):
|
def test_basic(self):
|
||||||
global_config = None
|
global_config = None
|
||||||
myconf = self.write_file('my.conf', '')
|
settings = {"wutta_config": self.config}
|
||||||
settings = {'wutta.config': myconf}
|
|
||||||
app = mod.main(global_config, **settings)
|
app = mod.main(global_config, **settings)
|
||||||
self.assertIsInstance(app, Router)
|
self.assertIsInstance(app, Router)
|
||||||
|
self.assertIsNone(CoreOpSession.session_factory.kw["bind"])
|
||||||
|
|
||||||
|
def test_corepos_engine(self):
|
||||||
|
engine = sa.create_engine("sqlite://")
|
||||||
|
self.config.core_office_op_engine = engine
|
||||||
|
settings = {"wutta_config": self.config}
|
||||||
|
app = mod.main(None, **settings)
|
||||||
|
self.assertIsInstance(app, Router)
|
||||||
|
self.assertIs(CoreOpSession.session_factory.kw["bind"], engine)
|
||||||
|
|
||||||
|
|
||||||
class TestMakeWsgiApp(ConfigTestCase):
|
class TestMakeWsgiApp(DataTestCase):
|
||||||
|
|
||||||
def test_basic(self):
|
def test_basic(self):
|
||||||
wsgi = mod.make_wsgi_app()
|
wsgi = mod.make_wsgi_app(config=self.config)
|
||||||
self.assertIsInstance(wsgi, Router)
|
self.assertIsInstance(wsgi, Router)
|
||||||
|
|
||||||
|
|
||||||
class TestMakeAsgiApp(ConfigTestCase):
|
class TestMakeAsgiApp(DataTestCase):
|
||||||
|
|
||||||
def test_basic(self):
|
def test_basic(self):
|
||||||
asgi = mod.make_asgi_app()
|
asgi = mod.make_asgi_app(config=self.config)
|
||||||
self.assertIsInstance(asgi, WsgiToAsgi)
|
self.assertIsInstance(asgi, WsgiToAsgi)
|
||||||
|
|
|
@ -13,39 +13,48 @@ class TestSideshowMenuHandler(WebTestCase):
|
||||||
def test_make_customers_menu(self):
|
def test_make_customers_menu(self):
|
||||||
handler = self.make_handler()
|
handler = self.make_handler()
|
||||||
menu = handler.make_customers_menu(self.request)
|
menu = handler.make_customers_menu(self.request)
|
||||||
item = menu['items'][-1]
|
item = menu["items"][-1]
|
||||||
self.assertEqual(item, {
|
self.assertEqual(
|
||||||
'title': "CORE-POS Members",
|
item,
|
||||||
'route': 'corepos_members',
|
{
|
||||||
'perm': 'corepos_members.list',
|
"title": "CORE-POS Members",
|
||||||
})
|
"route": "corepos_members",
|
||||||
|
"perm": "corepos_members.list",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
def test_make_products_menu(self):
|
def test_make_products_menu(self):
|
||||||
handler = self.make_handler()
|
handler = self.make_handler()
|
||||||
menu = handler.make_products_menu(self.request)
|
menu = handler.make_products_menu(self.request)
|
||||||
item = menu['items'][-1]
|
item = menu["items"][-1]
|
||||||
self.assertEqual(item, {
|
self.assertEqual(
|
||||||
'title': "CORE-POS Products",
|
item,
|
||||||
'route': 'corepos_products',
|
{
|
||||||
'perm': 'corepos_products.list',
|
"title": "CORE-POS Products",
|
||||||
})
|
"route": "corepos_products",
|
||||||
|
"perm": "corepos_products.list",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
def test_make_other_menu(self):
|
def test_make_other_menu(self):
|
||||||
handler = self.make_handler()
|
handler = self.make_handler()
|
||||||
|
|
||||||
# no url configured by default
|
# no url configured by default
|
||||||
menu = handler.make_other_menu(self.request)
|
menu = handler.make_other_menu(self.request)
|
||||||
if menu['items']:
|
if menu["items"]:
|
||||||
item = menu['items'][-1]
|
item = menu["items"][-1]
|
||||||
self.assertNotEqual(item['title'], "CORE Office")
|
self.assertNotEqual(item["title"], "CORE Office")
|
||||||
|
|
||||||
# entry added if url configured
|
# entry added if url configured
|
||||||
self.config.setdefault('corepos.office.url', 'http://localhost/fannie/')
|
self.config.setdefault("corepos.office.url", "http://localhost/fannie/")
|
||||||
menu = handler.make_other_menu(self.request)
|
menu = handler.make_other_menu(self.request)
|
||||||
item = menu['items'][-1]
|
item = menu["items"][-1]
|
||||||
self.assertEqual(item, {
|
self.assertEqual(
|
||||||
'title': "CORE Office",
|
item,
|
||||||
# nb. trailing slash gets stripped
|
{
|
||||||
'url': 'http://localhost/fannie',
|
"title": "CORE Office",
|
||||||
'target': '_blank',
|
# nb. trailing slash gets stripped
|
||||||
})
|
"url": "http://localhost/fannie",
|
||||||
|
"target": "_blank",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
4
tox.ini
4
tox.ini
|
@ -6,6 +6,10 @@ envlist = py38, py39, py310, py311
|
||||||
extras = tests
|
extras = tests
|
||||||
commands = pytest {posargs}
|
commands = pytest {posargs}
|
||||||
|
|
||||||
|
[testenv:pylint]
|
||||||
|
basepython = python3.11
|
||||||
|
commands = pylint sideshow_corepos
|
||||||
|
|
||||||
[testenv:coverage]
|
[testenv:coverage]
|
||||||
basepython = python3.11
|
basepython = python3.11
|
||||||
commands = pytest --cov=sideshow_corepos --cov-report=html --cov-fail-under=100
|
commands = pytest --cov=sideshow_corepos --cov-report=html --cov-fail-under=100
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue