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)
|
||||
|
||||
### Feat
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
|
||||
# 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/
|
||||
|
|
29
docs/conf.py
29
docs/conf.py
|
@ -8,33 +8,32 @@
|
|||
|
||||
from importlib.metadata import version as get_version
|
||||
|
||||
project = 'Sideshow-COREPOS'
|
||||
copyright = '2025, Lance Edgar'
|
||||
author = 'Lance Edgar'
|
||||
release = '0.1'
|
||||
release = get_version('Sideshow-COREPOS')
|
||||
project = "Sideshow-COREPOS"
|
||||
copyright = "2025, Lance Edgar"
|
||||
author = "Lance Edgar"
|
||||
release = get_version("Sideshow-COREPOS")
|
||||
|
||||
# -- General configuration ---------------------------------------------------
|
||||
# https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration
|
||||
|
||||
extensions = [
|
||||
'sphinx.ext.autodoc',
|
||||
'sphinx.ext.intersphinx',
|
||||
'sphinx.ext.viewcode',
|
||||
'sphinx.ext.todo',
|
||||
"sphinx.ext.autodoc",
|
||||
"sphinx.ext.intersphinx",
|
||||
"sphinx.ext.viewcode",
|
||||
"sphinx.ext.todo",
|
||||
]
|
||||
|
||||
templates_path = ['_templates']
|
||||
exclude_patterns = ['_build', 'Thumbs.db', '.DS_Store']
|
||||
templates_path = ["_templates"]
|
||||
exclude_patterns = ["_build", "Thumbs.db", ".DS_Store"]
|
||||
|
||||
intersphinx_mapping = {
|
||||
'sideshow': ('https://rattailproject.org/docs/sideshow/', None),
|
||||
'wuttjamaican': ('https://rattailproject.org/docs/wuttjamaican/', None),
|
||||
"sideshow": ("https://docs.wuttaproject.org/sideshow/", None),
|
||||
"wuttjamaican": ("https://docs.wuttaproject.org/wuttjamaican/", None),
|
||||
}
|
||||
|
||||
|
||||
# -- Options for HTML output -------------------------------------------------
|
||||
# https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output
|
||||
|
||||
html_theme = 'furo'
|
||||
html_static_path = ['_static']
|
||||
html_theme = "furo"
|
||||
html_static_path = ["_static"]
|
||||
|
|
|
@ -8,6 +8,12 @@ This is `Sideshow`_ with integration for `CORE-POS`_.
|
|||
|
||||
.. _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::
|
||||
:maxdepth: 1
|
||||
|
|
|
@ -5,7 +5,7 @@ build-backend = "hatchling.build"
|
|||
|
||||
[project]
|
||||
name = "Sideshow-COREPOS"
|
||||
version = "0.1.0"
|
||||
version = "0.1.1"
|
||||
description = "Case/Special Order Tracker for CORE-POS"
|
||||
readme = "README.md"
|
||||
authors = [
|
||||
|
@ -27,17 +27,19 @@ classifiers = [
|
|||
"Programming Language :: Python :: 3.9",
|
||||
"Programming Language :: Python :: 3.10",
|
||||
"Programming Language :: Python :: 3.11",
|
||||
"Topic :: Office/Business",
|
||||
"Topic :: Software Development :: Libraries :: Python Modules",
|
||||
]
|
||||
license = {text = "GNU General Public License v3+"}
|
||||
requires-python = ">= 3.8"
|
||||
dependencies = [
|
||||
"Sideshow>=0.3.0",
|
||||
"Wutta-COREPOS[web]>=0.2.0",
|
||||
"Sideshow>=0.7.1",
|
||||
"Wutta-COREPOS[web]>=0.3.0",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
docs = ["Sphinx", "furo"]
|
||||
tests = ["pytest-cov", "tox"]
|
||||
tests = ["pylint", "pytest", "pytest-cov", "tox"]
|
||||
|
||||
|
||||
[project.entry-points."paste.app_factory"]
|
||||
|
|
|
@ -43,28 +43,35 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
|
|||
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()
|
||||
op_model = corepos.get_model_office_op()
|
||||
op_session = corepos.make_session_office_op()
|
||||
|
||||
# base query
|
||||
query = op_session.query(op_model.CustomerClassic)\
|
||||
.join(op_model.MemberInfo,
|
||||
op_model.MemberInfo.card_number == op_model.CustomerClassic.card_number)
|
||||
query = op_session.query(op_model.CustomerClassic).join(
|
||||
op_model.MemberInfo,
|
||||
op_model.MemberInfo.card_number == op_model.CustomerClassic.card_number,
|
||||
)
|
||||
|
||||
# filter query
|
||||
criteria = []
|
||||
for word in term.split():
|
||||
criteria.append(sa.or_(
|
||||
op_model.CustomerClassic.first_name.ilike(f'%{word}%'),
|
||||
op_model.CustomerClassic.last_name.ilike(f'%{word}%')))
|
||||
criteria.append(
|
||||
sa.or_(
|
||||
op_model.CustomerClassic.first_name.ilike(f"%{word}%"),
|
||||
op_model.CustomerClassic.last_name.ilike(f"%{word}%"),
|
||||
)
|
||||
)
|
||||
query = query.filter(sa.and_(*criteria))
|
||||
|
||||
# sort query
|
||||
query = query.order_by(op_model.CustomerClassic.first_name,
|
||||
op_model.CustomerClassic.last_name)
|
||||
query = query.order_by(
|
||||
op_model.CustomerClassic.first_name, op_model.CustomerClassic.last_name
|
||||
)
|
||||
|
||||
# get data
|
||||
# TODO: need max_results option
|
||||
|
@ -72,32 +79,41 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
|
|||
|
||||
# get results
|
||||
def result(customer):
|
||||
return {'value': str(customer.card_number),
|
||||
'label': str(customer)}
|
||||
return {"value": str(customer.card_number), "label": str(customer)}
|
||||
|
||||
results = [result(c) for c in customers]
|
||||
|
||||
op_session.close()
|
||||
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()
|
||||
op_model = corepos.get_model_office_op()
|
||||
op_session = corepos.make_session_office_op()
|
||||
|
||||
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:
|
||||
customer = op_session.query(op_model.CustomerClassic)\
|
||||
.join(op_model.MemberInfo,
|
||||
op_model.MemberInfo.card_number == op_model.CustomerClassic.card_number)\
|
||||
.filter(op_model.CustomerClassic.card_number == int(batch.customer_id))\
|
||||
.filter(op_model.CustomerClassic.person_number == 1)\
|
||||
.options(orm.joinedload(op_model.CustomerClassic.member_info))\
|
||||
.one()
|
||||
except orm.exc.NoResultFound:
|
||||
raise ValueError(f"CORE-POS Customer not found: {batch.customer_id}")
|
||||
customer = (
|
||||
op_session.query(op_model.CustomerClassic)
|
||||
.join(
|
||||
op_model.MemberInfo,
|
||||
op_model.MemberInfo.card_number
|
||||
== op_model.CustomerClassic.card_number,
|
||||
)
|
||||
.filter(op_model.CustomerClassic.card_number == int(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.phone_number = customer.member_info.phone
|
||||
|
@ -105,7 +121,9 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
|
|||
|
||||
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()
|
||||
op_model = corepos.get_model_office_op()
|
||||
|
@ -117,14 +135,16 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
|
|||
# filter query
|
||||
criteria = []
|
||||
for word in term.split():
|
||||
criteria.append(sa.or_(
|
||||
op_model.Product.brand.ilike(f'%{word}%'),
|
||||
op_model.Product.description.ilike(f'%{word}%')))
|
||||
criteria.append(
|
||||
sa.or_(
|
||||
op_model.Product.brand.ilike(f"%{word}%"),
|
||||
op_model.Product.description.ilike(f"%{word}%"),
|
||||
)
|
||||
)
|
||||
query = query.filter(sa.and_(*criteria))
|
||||
|
||||
# sort query
|
||||
query = query.order_by(op_model.Product.brand,
|
||||
op_model.Product.description)
|
||||
query = query.order_by(op_model.Product.brand, op_model.Product.description)
|
||||
|
||||
# get data
|
||||
# TODO: need max_results option
|
||||
|
@ -132,43 +152,50 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
|
|||
|
||||
# get results
|
||||
def result(product):
|
||||
return {'value': product.upc,
|
||||
'label': self.app.make_full_name(product.brand,
|
||||
product.description,
|
||||
product.size)}
|
||||
return {
|
||||
"value": product.upc,
|
||||
"label": self.app.make_full_name(
|
||||
product.brand, product.description, product.size
|
||||
),
|
||||
}
|
||||
|
||||
results = [result(c) for c in products]
|
||||
|
||||
op_session.close()
|
||||
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()
|
||||
op_model = corepos.get_model_office_op()
|
||||
op_session = corepos.make_session_office_op()
|
||||
|
||||
try:
|
||||
product = op_session.query(op_model.Product)\
|
||||
.filter(op_model.Product.upc == product_id)\
|
||||
.one()
|
||||
except orm.exc.NoResultFound:
|
||||
raise ValueError(f"CORE-POS Product not found: {product_id}")
|
||||
product = (
|
||||
op_session.query(op_model.Product)
|
||||
.filter(op_model.Product.upc == product_id)
|
||||
.one()
|
||||
)
|
||||
except orm.exc.NoResultFound as e:
|
||||
raise ValueError(f"CORE-POS Product not found: {product_id}") from e
|
||||
|
||||
data = {
|
||||
'product_id': product.upc,
|
||||
'scancode': product.upc,
|
||||
'brand_name': product.brand,
|
||||
'description': product.description,
|
||||
'size': product.size,
|
||||
'full_description': self.app.make_full_name(product.brand,
|
||||
product.description,
|
||||
product.size),
|
||||
'weighed': product.scale,
|
||||
'special_order': False,
|
||||
'department_id': product.department_number,
|
||||
'department_name': product.department.name if product.department else None,
|
||||
'case_size': self.get_case_size_for_external_product(product),
|
||||
'unit_price_reg': self.get_unit_price_reg_for_external_product(product),
|
||||
"product_id": product.upc,
|
||||
"scancode": product.upc,
|
||||
"brand_name": product.brand,
|
||||
"description": product.description,
|
||||
"size": product.size,
|
||||
"full_description": self.app.make_full_name(
|
||||
product.brand, product.description, product.size
|
||||
),
|
||||
"weighed": product.scale,
|
||||
"special_order": False,
|
||||
"department_id": product.department_number,
|
||||
"department_name": product.department.name if product.department else None,
|
||||
"case_size": self.get_case_size_for_external_product(product),
|
||||
"unit_price_reg": self.get_unit_price_reg_for_external_product(product),
|
||||
# TODO
|
||||
# 'vendor_name': product.vendor_name,
|
||||
# 'vendor_item_code': product.vendor_item_code,
|
||||
|
@ -177,18 +204,20 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
|
|||
op_session.close()
|
||||
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()
|
||||
op_model = corepos.get_model_office_op()
|
||||
op_session = corepos.make_session_office_op()
|
||||
|
||||
try:
|
||||
product = op_session.query(op_model.Product)\
|
||||
.filter(op_model.Product.upc == row.product_id)\
|
||||
.one()
|
||||
except orm.exc.NoResultFound:
|
||||
raise ValueError(f"CORE-POS Product not found: {row.product_id}")
|
||||
product = (
|
||||
op_session.query(op_model.Product)
|
||||
.filter(op_model.Product.upc == row.product_id)
|
||||
.one()
|
||||
)
|
||||
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_brand = product.brand
|
||||
|
@ -198,20 +227,34 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
|
|||
row.department_id = product.department_number
|
||||
row.department_name = product.department.name if product.department else None
|
||||
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.unit_cost = product.cost
|
||||
row.unit_price_reg = self.get_unit_price_reg_for_external_product(product)
|
||||
|
||||
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:
|
||||
item = product.vendor_items[0]
|
||||
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:
|
||||
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.
|
||||
"""
|
||||
key = 'sideshow_corepos'
|
||||
|
||||
def configure(self, config):
|
||||
key = "sideshow_corepos"
|
||||
|
||||
def configure(self, config): # pylint: disable=empty-docstring
|
||||
""" """
|
||||
|
||||
# batch handlers
|
||||
config.setdefault(f'{config.appname}.batch.neworder.handler.spec',
|
||||
'sideshow_corepos.batch.neworder:NewOrderBatchHandler')
|
||||
config.setdefault(
|
||||
f"{config.appname}.batch.neworder.handler.spec",
|
||||
"sideshow_corepos.batch.neworder:NewOrderBatchHandler",
|
||||
)
|
||||
|
||||
# web app menu
|
||||
config.setdefault(f'{config.appname}.web.menus.handler.spec',
|
||||
'sideshow_corepos.web.menus:SideshowMenuHandler')
|
||||
config.setdefault(
|
||||
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):
|
||||
config.include('sideshow_corepos.web.views')
|
||||
def includeme(config): # pylint: disable=missing-function-docstring
|
||||
config.include("sideshow_corepos.web.views")
|
||||
|
|
|
@ -29,39 +29,43 @@ from wuttaweb import app as base
|
|||
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).
|
||||
"""
|
||||
# prefer Sideshow templates over wuttaweb
|
||||
settings.setdefault('mako.directories', [
|
||||
'sideshow.web:templates',
|
||||
'wuttaweb:templates',
|
||||
])
|
||||
settings.setdefault(
|
||||
"mako.directories",
|
||||
[
|
||||
"sideshow.web:templates",
|
||||
"wuttaweb:templates",
|
||||
],
|
||||
)
|
||||
|
||||
# make config objects
|
||||
wutta_config = base.make_wutta_config(settings)
|
||||
pyramid_config = base.make_pyramid_config(settings)
|
||||
|
||||
# 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
|
||||
pyramid_config.include('sideshow.web')
|
||||
pyramid_config.include('sideshow_corepos.web')
|
||||
pyramid_config.include("sideshow.web")
|
||||
pyramid_config.include("sideshow_corepos.web")
|
||||
|
||||
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).
|
||||
"""
|
||||
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).
|
||||
"""
|
||||
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['items'].extend([
|
||||
{'type': 'sep'},
|
||||
{
|
||||
'title': "CORE-POS Members",
|
||||
'route': 'corepos_members',
|
||||
'perm': 'corepos_members.list',
|
||||
},
|
||||
])
|
||||
menu["items"].extend(
|
||||
[
|
||||
{"type": "sep"},
|
||||
{
|
||||
"title": "CORE-POS Members",
|
||||
"route": "corepos_members",
|
||||
"perm": "corepos_members.list",
|
||||
},
|
||||
]
|
||||
)
|
||||
|
||||
return menu
|
||||
|
||||
|
@ -55,14 +57,16 @@ class SideshowMenuHandler(base.SideshowMenuHandler):
|
|||
"""
|
||||
menu = super().make_products_menu(request, **kwargs)
|
||||
|
||||
menu['items'].extend([
|
||||
{'type': 'sep'},
|
||||
{
|
||||
'title': "CORE-POS Products",
|
||||
'route': 'corepos_products',
|
||||
'perm': 'corepos_products.list',
|
||||
},
|
||||
])
|
||||
menu["items"].extend(
|
||||
[
|
||||
{"type": "sep"},
|
||||
{
|
||||
"title": "CORE-POS Products",
|
||||
"route": "corepos_products",
|
||||
"perm": "corepos_products.list",
|
||||
},
|
||||
]
|
||||
)
|
||||
|
||||
return menu
|
||||
|
||||
|
@ -75,12 +79,14 @@ class SideshowMenuHandler(base.SideshowMenuHandler):
|
|||
corepos = self.app.get_corepos_handler()
|
||||
url = corepos.get_office_url()
|
||||
if url:
|
||||
menu['items'].extend([
|
||||
{
|
||||
'title': "CORE Office",
|
||||
'url': url,
|
||||
'target': '_blank',
|
||||
},
|
||||
])
|
||||
menu["items"].extend(
|
||||
[
|
||||
{
|
||||
"title": "CORE Office",
|
||||
"url": url,
|
||||
"target": "_blank",
|
||||
},
|
||||
]
|
||||
)
|
||||
|
||||
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
|
||||
config.include('wutta_corepos.web.views.corepos.members')
|
||||
config.include('wutta_corepos.web.views.corepos.products')
|
||||
config.include("wutta_corepos.web.views.corepos.members")
|
||||
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
|
||||
"""
|
||||
if not skip_tests:
|
||||
c.run('pytest')
|
||||
c.run("pytest")
|
||||
|
||||
# rebuild pkg
|
||||
if os.path.exists('dist'):
|
||||
shutil.rmtree('dist')
|
||||
if os.path.exists('Sideshow_COREPOS.egg-info'):
|
||||
shutil.rmtree('Sideshow_COREPOS.egg-info')
|
||||
c.run('python -m build --sdist')
|
||||
if os.path.exists("dist"):
|
||||
shutil.rmtree("dist")
|
||||
if os.path.exists("Sideshow_COREPOS.egg-info"):
|
||||
shutil.rmtree("Sideshow_COREPOS.egg-info")
|
||||
c.run("python -m build --sdist")
|
||||
|
||||
# upload
|
||||
c.run('twine upload dist/*')
|
||||
c.run("twine upload dist/*")
|
||||
|
|
|
@ -17,8 +17,8 @@ class TestNewOrderBatchHandler(DataTestCase):
|
|||
def setUp(self):
|
||||
super().setUp()
|
||||
|
||||
self.op_engine = sa.create_engine('sqlite://')
|
||||
self.config.core_office_op_engines = {'default': self.op_engine}
|
||||
self.op_engine = sa.create_engine("sqlite://")
|
||||
self.config.core_office_op_engines = {"default": self.op_engine}
|
||||
self.config.core_office_op_engine = 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):
|
||||
config = super().make_config(**kwargs)
|
||||
config.setdefault('wutta.enum_spec', 'sideshow.enum')
|
||||
config.setdefault("wutta.enum_spec", "sideshow.enum")
|
||||
return config
|
||||
|
||||
def make_handler(self):
|
||||
|
@ -41,33 +41,41 @@ class TestNewOrderBatchHandler(DataTestCase):
|
|||
handler = self.make_handler()
|
||||
|
||||
# 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
|
||||
member = op_model.MemberInfo(card_number=42)
|
||||
self.op_session.add(member)
|
||||
customer = op_model.CustomerClassic(first_name="Chuck", last_name="Norris",
|
||||
last_change=datetime.datetime.now())
|
||||
customer = op_model.CustomerClassic(
|
||||
first_name="Chuck", last_name="Norris", last_change=datetime.datetime.now()
|
||||
)
|
||||
member.customers.append(customer)
|
||||
self.op_session.add(customer)
|
||||
self.op_session.flush()
|
||||
|
||||
# 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(results[0], {
|
||||
'value': '42',
|
||||
'label': "Chuck Norris",
|
||||
})
|
||||
self.assertEqual(
|
||||
results[0],
|
||||
{
|
||||
"value": "42",
|
||||
"label": "Chuck Norris",
|
||||
},
|
||||
)
|
||||
|
||||
# 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):
|
||||
model = self.app.model
|
||||
handler = self.make_handler()
|
||||
|
||||
user = model.User(username='barney')
|
||||
user = model.User(username="barney")
|
||||
self.session.add(user)
|
||||
self.session.flush()
|
||||
|
||||
|
@ -76,91 +84,114 @@ class TestNewOrderBatchHandler(DataTestCase):
|
|||
self.session.flush()
|
||||
|
||||
# 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)
|
||||
customer = op_model.CustomerClassic(first_name="Chuck", last_name="Norris",
|
||||
last_change=datetime.datetime.now())
|
||||
customer = op_model.CustomerClassic(
|
||||
first_name="Chuck", last_name="Norris", last_change=datetime.datetime.now()
|
||||
)
|
||||
member.customers.append(customer)
|
||||
self.op_session.add(customer)
|
||||
self.op_session.flush()
|
||||
|
||||
# error if invalid customer_id
|
||||
batch.customer_id = 'BreakThings!'
|
||||
self.assertRaises(ValueError, handler.refresh_batch_from_external_customer, batch)
|
||||
batch.customer_id = "BreakThings!"
|
||||
self.assertRaises(
|
||||
ValueError, handler.refresh_batch_from_external_customer, batch
|
||||
)
|
||||
|
||||
# error if customer not found
|
||||
batch.customer_id = '9999'
|
||||
self.assertRaises(ValueError, handler.refresh_batch_from_external_customer, batch)
|
||||
batch.customer_id = "9999"
|
||||
self.assertRaises(
|
||||
ValueError, handler.refresh_batch_from_external_customer, batch
|
||||
)
|
||||
|
||||
# batch should reflect customer info
|
||||
batch.customer_id = '42'
|
||||
batch.customer_id = "42"
|
||||
self.assertIsNone(batch.customer_name)
|
||||
self.assertIsNone(batch.phone_number)
|
||||
self.assertIsNone(batch.email_address)
|
||||
handler.refresh_batch_from_external_customer(batch)
|
||||
self.assertEqual(batch.customer_name, "Chuck Norris")
|
||||
self.assertEqual(batch.phone_number, '555-1234')
|
||||
self.assertEqual(batch.email_address, 'chuck@example.com')
|
||||
self.assertEqual(batch.phone_number, "555-1234")
|
||||
self.assertEqual(batch.email_address, "chuck@example.com")
|
||||
|
||||
def test_autocomplete_products_local(self):
|
||||
handler = self.make_handler()
|
||||
|
||||
# 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
|
||||
product = op_model.Product(upc='07430500132', brand="Bragg's",
|
||||
description="Vinegar", size='32oz')
|
||||
product = op_model.Product(
|
||||
upc="07430500132", brand="Bragg's", description="Vinegar", size="32oz"
|
||||
)
|
||||
self.op_session.add(product)
|
||||
self.op_session.commit()
|
||||
|
||||
# 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(results[0], {
|
||||
'value': '07430500132',
|
||||
'label': "Bragg's Vinegar 32oz",
|
||||
})
|
||||
self.assertEqual(
|
||||
results[0],
|
||||
{
|
||||
"value": "07430500132",
|
||||
"label": "Bragg's Vinegar 32oz",
|
||||
},
|
||||
)
|
||||
|
||||
# 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(results[0], {
|
||||
'value': '07430500132',
|
||||
'label': "Bragg's Vinegar 32oz",
|
||||
})
|
||||
self.assertEqual(
|
||||
results[0],
|
||||
{
|
||||
"value": "07430500132",
|
||||
"label": "Bragg's Vinegar 32oz",
|
||||
},
|
||||
)
|
||||
|
||||
# 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):
|
||||
handler = self.make_handler()
|
||||
|
||||
# null
|
||||
product = op_model.Product(upc='07430500132', brand="Bragg's",
|
||||
description="Vinegar", size='32oz')
|
||||
product = op_model.Product(
|
||||
upc="07430500132", brand="Bragg's", description="Vinegar", size="32oz"
|
||||
)
|
||||
self.op_session.add(product)
|
||||
self.op_session.commit()
|
||||
self.op_session.refresh(product)
|
||||
self.assertIsNone(handler.get_case_size_for_external_product(product))
|
||||
|
||||
# typical
|
||||
vendor = op_model.Vendor(id=42, name='Acme Distributors')
|
||||
vendor = op_model.Vendor(id=42, name="Acme Distributors")
|
||||
self.op_session.add(vendor)
|
||||
item = op_model.VendorItem(vendor=vendor, sku='1234', units=12.34,
|
||||
vendor_item_id=1)
|
||||
item = op_model.VendorItem(
|
||||
vendor=vendor, sku="1234", units=12.34, vendor_item_id=1
|
||||
)
|
||||
product.vendor_items.append(item)
|
||||
self.op_session.commit()
|
||||
self.op_session.refresh(product)
|
||||
self.assertEqual(handler.get_case_size_for_external_product(product),
|
||||
decimal.Decimal('12.3400'))
|
||||
self.assertEqual(
|
||||
handler.get_case_size_for_external_product(product),
|
||||
decimal.Decimal("12.3400"),
|
||||
)
|
||||
|
||||
def test_get_unit_price_reg_for_external_product(self):
|
||||
handler = self.make_handler()
|
||||
|
||||
# null
|
||||
product = op_model.Product(upc='07430500132', brand="Bragg's",
|
||||
description="Vinegar", size='32oz')
|
||||
product = op_model.Product(
|
||||
upc="07430500132", brand="Bragg's", description="Vinegar", size="32oz"
|
||||
)
|
||||
self.op_session.add(product)
|
||||
self.op_session.commit()
|
||||
self.op_session.refresh(product)
|
||||
|
@ -170,53 +201,64 @@ class TestNewOrderBatchHandler(DataTestCase):
|
|||
product.normal_price = 4.19
|
||||
self.op_session.commit()
|
||||
self.op_session.refresh(product)
|
||||
self.assertEqual(handler.get_unit_price_reg_for_external_product(product),
|
||||
decimal.Decimal('4.19'))
|
||||
self.assertEqual(
|
||||
handler.get_unit_price_reg_for_external_product(product),
|
||||
decimal.Decimal("4.19"),
|
||||
)
|
||||
|
||||
def test_get_product_info_external(self):
|
||||
model = self.app.model
|
||||
handler = self.make_handler()
|
||||
|
||||
user = model.User(username='barney')
|
||||
user = model.User(username="barney")
|
||||
self.session.add(user)
|
||||
batch = handler.make_batch(self.session, created_by=user)
|
||||
self.session.add(batch)
|
||||
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)
|
||||
product = op_model.Product(upc='07430500132', brand="Bragg",
|
||||
description="Vinegar", size='32oz',
|
||||
normal_price=4.19)
|
||||
item = op_model.VendorItem(vendor=vendor, sku='1234', units=12.34,
|
||||
vendor_item_id=1)
|
||||
product = op_model.Product(
|
||||
upc="07430500132",
|
||||
brand="Bragg",
|
||||
description="Vinegar",
|
||||
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)
|
||||
self.op_session.add(product)
|
||||
self.op_session.commit()
|
||||
|
||||
# typical
|
||||
info = handler.get_product_info_external(self.session, '07430500132')
|
||||
self.assertEqual(info['product_id'], '07430500132')
|
||||
self.assertEqual(info['scancode'], '07430500132')
|
||||
self.assertEqual(info['brand_name'], 'Bragg')
|
||||
self.assertEqual(info['description'], 'Vinegar')
|
||||
self.assertEqual(info['size'], '32oz')
|
||||
self.assertEqual(info['full_description'], 'Bragg Vinegar 32oz')
|
||||
self.assertEqual(info['case_size'], decimal.Decimal('12.3400'))
|
||||
self.assertEqual(info['unit_price_reg'], decimal.Decimal('4.19'))
|
||||
info = handler.get_product_info_external(self.session, "07430500132")
|
||||
self.assertEqual(info["product_id"], "07430500132")
|
||||
self.assertEqual(info["scancode"], "07430500132")
|
||||
self.assertEqual(info["brand_name"], "Bragg")
|
||||
self.assertEqual(info["description"], "Vinegar")
|
||||
self.assertEqual(info["size"], "32oz")
|
||||
self.assertEqual(info["full_description"], "Bragg Vinegar 32oz")
|
||||
self.assertEqual(info["case_size"], decimal.Decimal("12.3400"))
|
||||
self.assertEqual(info["unit_price_reg"], decimal.Decimal("4.19"))
|
||||
|
||||
# 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
|
||||
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):
|
||||
model = self.app.model
|
||||
enum = self.app.enum
|
||||
handler = self.make_handler()
|
||||
|
||||
user = model.User(username='barney')
|
||||
user = model.User(username="barney")
|
||||
self.session.add(user)
|
||||
batch = handler.make_batch(self.session, created_by=user)
|
||||
self.session.add(batch)
|
||||
|
@ -225,27 +267,32 @@ class TestNewOrderBatchHandler(DataTestCase):
|
|||
self.session.add(row)
|
||||
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)
|
||||
product = op_model.Product(upc='07430500132', brand="Bragg",
|
||||
description="Vinegar", size='32oz',
|
||||
normal_price=4.19)
|
||||
item = op_model.VendorItem(vendor=vendor, sku='1234', units=12.34,
|
||||
vendor_item_id=1)
|
||||
product = op_model.Product(
|
||||
upc="07430500132",
|
||||
brand="Bragg",
|
||||
description="Vinegar",
|
||||
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)
|
||||
self.op_session.add(product)
|
||||
self.op_session.commit()
|
||||
|
||||
# error if invalid product_id
|
||||
row.product_id = 'BreakThings!'
|
||||
row.product_id = "BreakThings!"
|
||||
self.assertRaises(ValueError, handler.refresh_row_from_external_product, row)
|
||||
|
||||
# error if product not found
|
||||
row.product_id = '9999'
|
||||
row.product_id = "9999"
|
||||
self.assertRaises(ValueError, handler.refresh_row_from_external_product, row)
|
||||
|
||||
# row should reflect product info
|
||||
row.product_id = '07430500132'
|
||||
row.product_id = "07430500132"
|
||||
self.assertIsNone(row.product_scancode)
|
||||
self.assertIsNone(row.product_brand)
|
||||
self.assertIsNone(row.product_description)
|
||||
|
@ -253,9 +300,11 @@ class TestNewOrderBatchHandler(DataTestCase):
|
|||
self.assertIsNone(row.case_size)
|
||||
self.assertIsNone(row.unit_price_reg)
|
||||
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_description, "Vinegar")
|
||||
self.assertEqual(row.product_size, "32oz")
|
||||
self.assertEqual(row.case_size, decimal.Decimal('12.3400'))
|
||||
self.assertEqual(row.unit_price_reg, decimal.Decimal('4.19'))
|
||||
self.assertEqual(row.case_size, decimal.Decimal("12.3400"))
|
||||
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; -*-
|
||||
|
||||
from wuttjamaican.testing import FileTestCase, ConfigTestCase
|
||||
import sqlalchemy as sa
|
||||
|
||||
from wuttjamaican.testing import DataTestCase
|
||||
|
||||
from asgiref.wsgi import WsgiToAsgi
|
||||
from pyramid.router import Router
|
||||
|
||||
from wutta_corepos.web.db import CoreOpSession
|
||||
|
||||
from sideshow_corepos.web import app as mod
|
||||
|
||||
|
||||
class TestMain(FileTestCase):
|
||||
class TestMain(DataTestCase):
|
||||
|
||||
def test_basic(self):
|
||||
global_config = None
|
||||
myconf = self.write_file('my.conf', '')
|
||||
settings = {'wutta.config': myconf}
|
||||
settings = {"wutta_config": self.config}
|
||||
app = mod.main(global_config, **settings)
|
||||
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):
|
||||
wsgi = mod.make_wsgi_app()
|
||||
wsgi = mod.make_wsgi_app(config=self.config)
|
||||
self.assertIsInstance(wsgi, Router)
|
||||
|
||||
|
||||
class TestMakeAsgiApp(ConfigTestCase):
|
||||
class TestMakeAsgiApp(DataTestCase):
|
||||
|
||||
def test_basic(self):
|
||||
asgi = mod.make_asgi_app()
|
||||
asgi = mod.make_asgi_app(config=self.config)
|
||||
self.assertIsInstance(asgi, WsgiToAsgi)
|
||||
|
|
|
@ -13,39 +13,48 @@ class TestSideshowMenuHandler(WebTestCase):
|
|||
def test_make_customers_menu(self):
|
||||
handler = self.make_handler()
|
||||
menu = handler.make_customers_menu(self.request)
|
||||
item = menu['items'][-1]
|
||||
self.assertEqual(item, {
|
||||
'title': "CORE-POS Members",
|
||||
'route': 'corepos_members',
|
||||
'perm': 'corepos_members.list',
|
||||
})
|
||||
item = menu["items"][-1]
|
||||
self.assertEqual(
|
||||
item,
|
||||
{
|
||||
"title": "CORE-POS Members",
|
||||
"route": "corepos_members",
|
||||
"perm": "corepos_members.list",
|
||||
},
|
||||
)
|
||||
|
||||
def test_make_products_menu(self):
|
||||
handler = self.make_handler()
|
||||
menu = handler.make_products_menu(self.request)
|
||||
item = menu['items'][-1]
|
||||
self.assertEqual(item, {
|
||||
'title': "CORE-POS Products",
|
||||
'route': 'corepos_products',
|
||||
'perm': 'corepos_products.list',
|
||||
})
|
||||
item = menu["items"][-1]
|
||||
self.assertEqual(
|
||||
item,
|
||||
{
|
||||
"title": "CORE-POS Products",
|
||||
"route": "corepos_products",
|
||||
"perm": "corepos_products.list",
|
||||
},
|
||||
)
|
||||
|
||||
def test_make_other_menu(self):
|
||||
handler = self.make_handler()
|
||||
|
||||
# no url configured by default
|
||||
menu = handler.make_other_menu(self.request)
|
||||
if menu['items']:
|
||||
item = menu['items'][-1]
|
||||
self.assertNotEqual(item['title'], "CORE Office")
|
||||
if menu["items"]:
|
||||
item = menu["items"][-1]
|
||||
self.assertNotEqual(item["title"], "CORE Office")
|
||||
|
||||
# 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)
|
||||
item = menu['items'][-1]
|
||||
self.assertEqual(item, {
|
||||
'title': "CORE Office",
|
||||
# nb. trailing slash gets stripped
|
||||
'url': 'http://localhost/fannie',
|
||||
'target': '_blank',
|
||||
})
|
||||
item = menu["items"][-1]
|
||||
self.assertEqual(
|
||||
item,
|
||||
{
|
||||
"title": "CORE Office",
|
||||
# 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
|
||||
commands = pytest {posargs}
|
||||
|
||||
[testenv:pylint]
|
||||
basepython = python3.11
|
||||
commands = pylint sideshow_corepos
|
||||
|
||||
[testenv:coverage]
|
||||
basepython = python3.11
|
||||
commands = pytest --cov=sideshow_corepos --cov-report=html --cov-fail-under=100
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue