wutta/wuttaweb
3
0
Fork 0
Code Issues Activity

feat: basic support for displaying version history

Browse source 
this is not terribly feature-rich yet, just the basics
This commit is contained in:
Lance Edgar 2025-10-29 18:32:35 -05:00
parent 6d2eccd0ea
commit f33448f64a
18 changed files with 1323 additions and 66 deletions
Download patch file Download diff file Expand all files Collapse all files

6
docs/api/wuttaweb.diffs.rst Normal file
View file

@ -0,0 +1,6 @@
``wuttaweb.diffs``
==================
.. automodule:: wuttaweb.diffs
:members:

1
docs/index.rst
View file

@ -41,6 +41,7 @@ the narrative docs are pretty scant. That will eventually change.
api/wuttaweb.db
api/wuttaweb.db.continuum
api/wuttaweb.db.sess
api/wuttaweb.diffs
api/wuttaweb.emails
api/wuttaweb.forms
api/wuttaweb.forms.base

8
pyproject.toml
View file

@ -50,7 +50,7 @@ dependencies = [
[project.optional-dependencies]
continuum = ["Wutta-Continuum"]
continuum = ["Wutta-Continuum>=0.2.1"]
docs = ["Sphinx", "furo", "sphinxcontrib-programoutput"]
tests = ["pylint", "pytest", "pytest-cov", "tox"]
@ -91,3 +91,9 @@ update_changelog_on_bump = true
exclude = [
"htmlcov/",
]
[tool.pytest.ini_options]
markers = [
"versioned: tests with SQLAlchemy-Continuum versioning feature enabled",
]

224
src/wuttaweb/diffs.py Normal file
View file

@ -0,0 +1,224 @@
# -*- coding: utf-8; -*-
################################################################################
#
# wuttaweb -- Web App for Wutta Framework
# Copyright © 2024-2025 Lance Edgar
#
# This file is part of Wutta Framework.
#
# Wutta Framework is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the Free
# Software Foundation, either version 3 of the License, or (at your option) any
# later version.
#
# Wutta Framework is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# Wutta Framework. If not, see <http://www.gnu.org/licenses/>.
#
################################################################################
"""
Tools for displaying simple data diffs
"""
import sqlalchemy as sa
from pyramid.renderers import render
from webhelpers2.html import HTML
class Diff:
"""
Represent / display a basic "diff" between two data records.
You must provide both the "old" and "new" data records, when
constructing an instance of this class. Then call
:meth:`render_html()` to display the diff table.
:param old_data: Dict of "old" data record.
:param new_data: Dict of "new" data record.
:param fields: Optional list of field names. If not specified,
will be derived from the data records.
:param nature: What sort of diff is being represented; must be one
of: ``("create", "update", "delete")``
:param old_color: Background color to display for "old/deleted"
field data, when applicable.
:param new_color: Background color to display for "new/created"
field data, when applicable.
"""
def __init__( # pylint: disable=too-many-arguments,too-many-positional-arguments
self,
old_data: dict,
new_data: dict,
fields: list = None,
nature="update",
old_color="#ffebe9",
new_color="#dafbe1",
):
self.old_data = old_data
self.new_data = new_data
self.columns = ["field name", "old value", "new value"]
self.fields = fields or self.make_fields()
self.nature = nature
self.old_color = old_color
self.new_color = new_color
def make_fields(self): # pylint: disable=missing-function-docstring
return sorted(set(self.old_data) | set(self.new_data), key=lambda x: x.lower())
def old_value(self, field): # pylint: disable=missing-function-docstring
return self.old_data.get(field)
def new_value(self, field): # pylint: disable=missing-function-docstring
return self.new_data.get(field)
def values_differ(self, field): # pylint: disable=missing-function-docstring
return self.new_value(field) != self.old_value(field)
def render_html(self, template="/diff.mako", **kwargs):
"""
Render the diff as HTML table.
:param template: Name of template to render, if you need to
override the default.
:param \\**kwargs: Remaining kwargs are passed as context to
the template renderer.
:returns: HTML literal string
"""
context = kwargs
context["diff"] = self
return HTML.literal(render(template, context))
def render_field_row(self, field): # pylint: disable=missing-function-docstring
is_diff = self.values_differ(field)
td_field = HTML.tag("td", class_="field", c=field)
td_old_value = HTML.tag(
"td",
c=self.render_old_value(field),
**self.get_old_value_attrs(is_diff),
)
td_new_value = HTML.tag(
"td",
c=self.render_new_value(field),
**self.get_new_value_attrs(is_diff),
)
return HTML.tag("tr", c=[td_field, td_old_value, td_new_value])
def render_old_value(self, field): # pylint: disable=missing-function-docstring
value = self.old_value(field)
return repr(value)
def render_new_value(self, field): # pylint: disable=missing-function-docstring
value = self.new_value(field)
return repr(value)
def get_old_value_attrs( # pylint: disable=missing-function-docstring
self, is_diff
):
attrs = {}
if self.nature == "update" and is_diff:
attrs["style"] = f"background-color: {self.old_color};"
elif self.nature == "delete":
attrs["style"] = f"background-color: {self.old_color};"
return attrs
def get_new_value_attrs( # pylint: disable=missing-function-docstring
self, is_diff
):
attrs = {}
if self.nature == "create":
attrs["style"] = f"background-color: {self.new_color};"
elif self.nature == "update" and is_diff:
attrs["style"] = f"background-color: {self.new_color};"
return attrs
class VersionDiff(Diff):
"""
Special diff class, for use with version history views. Note that
while based on :class:`Diff`, this class uses a different
signature for the constructor.
:param version: Reference to a Continuum version record (object).
:param \\**kwargs: Remaining kwargs are passed as-is to the
:class:`Diff` constructor.
"""
def __init__(self, version, **kwargs):
import sqlalchemy_continuum as continuum # pylint: disable=import-outside-toplevel
from wutta_continuum.util import ( # pylint: disable=import-outside-toplevel
render_operation_type,
)
self.version = version
self.model_class = continuum.parent_class(type(self.version))
self.mapper = sa.inspect(self.model_class)
self.version_mapper = sa.inspect(type(self.version))
self.title = kwargs.pop("title", self.model_class.__name__)
self.operation_title = render_operation_type(self.version.operation_type)
if "nature" not in kwargs:
if (
version.previous
and version.operation_type == continuum.Operation.DELETE
):
kwargs["nature"] = "delete"
elif version.previous:
kwargs["nature"] = "update"
else:
kwargs["nature"] = "create"
if "fields" not in kwargs:
kwargs["fields"] = self.get_default_fields()
old_data = {}
new_data = {}
for field in kwargs["fields"]:
if version.previous:
old_data[field] = getattr(version.previous, field)
new_data[field] = getattr(version, field)
super().__init__(old_data, new_data, **kwargs)
def get_default_fields(self): # pylint: disable=missing-function-docstring
fields = sorted(self.version_mapper.columns.keys())
unwanted = [
"transaction_id",
"end_transaction_id",
"operation_type",
]
return [field for field in fields if field not in unwanted]
def render_version_value(self, value): # pylint: disable=missing-function-docstring
return HTML.tag("span", c=[repr(value)], style="font-family: monospace;")
def render_old_value(self, field):
if self.nature == "create":
return ""
value = self.old_value(field)
return self.render_version_value(value)
def render_new_value(self, field):
if self.nature == "delete":
return ""
value = self.new_value(field)
return self.render_version_value(value)

35
src/wuttaweb/templates/base.mako
View file

@ -152,6 +152,8 @@
font-size: 2rem;
font-weight: bold;
margin-bottom: 0 !important;
display: flex;
gap: 0.6rem;
}
#content-title h1 {
@ -242,7 +244,9 @@
## nb. this is the index title proper
<div class="level-left">
<div id="header-index-title" class="level-item">
% if index_title:
% if index_title_rendered is not Undefined and index_title_rendered:
<h1 class="title">${index_title_rendered}</h1>
% elif index_title:
% if index_url:
<h1 class="title">${h.link_to(index_title, index_url)}</h1>
% else:
@ -279,7 +283,7 @@
class="has-background-primary">
<div style="display: flex; align-items: center; gap: 0.5rem; padding: 0.5rem;">
<div style="width: 60%; display: flex; gap: 0.5rem;">
<div style="width: 60%; display: flex; gap: 1rem;">
<h1 class="title has-text-white"
v-html="contentTitleHTML">
@ -781,7 +785,32 @@
% endif
</%def>
<%def name="render_prevnext_header_buttons()"></%def>
<%def name="render_prevnext_header_buttons()">
% if show_prev_next is not Undefined and show_prev_next:
<b-button tag="a"
% if prev_url:
href="${prev_url}"
% else:
href="#"
disabled
% endif
icon-pack="fas"
icon-left="arrow-left">
Older
</b-button>
<b-button tag="a"
% if next_url:
href="${next_url}"
% else:
href="#"
disabled
% endif
icon-pack="fas"
icon-left="arrow-right">
Newer
</b-button>
% endif
</%def>
##############################
## vue components + app

15
src/wuttaweb/templates/diff.mako Normal file
View file

@ -0,0 +1,15 @@
## -*- coding: utf-8; -*-
<table class="table is-fullwidth is-bordered is-narrow">
<thead>
<tr>
% for column in diff.columns:
<th>${column}</th>
% endfor
</tr>
</thead>
<tbody>
% for field in diff.fields:
${diff.render_field_row(field)}
% endfor
</tbody>
</table>

12
src/wuttaweb/templates/master/view.mako
View file

@ -5,6 +5,18 @@
<%def name="content_title()">${instance_title}</%def>
<%def name="render_instance_header_title_extras()">
${parent.render_instance_header_title_extras()}
% if master.should_expose_versions():
<b-button tag="a"
href="${master.get_action_url('versions', instance)}"
icon-pack="fas"
icon-left="history">
View History
</b-button>
% endif
</%def>
<%def name="page_layout()">
% if master.has_rows:

35
src/wuttaweb/templates/master/view_version.mako Normal file
View file

@ -0,0 +1,35 @@
## -*- coding: utf-8; -*-
<%inherit file="/master/form.mako" />
<%def name="title()">${index_title} » ${instance_title} » changes @ TXN ${transaction.id}</%def>
<%def name="content_title()">changes @ TXN ${transaction.id}</%def>
<%def name="page_content()">
<div class="wutta-form-wrapper">
<b-field label="Changed" horizontal>
<span>${changed}</span>
</b-field>
<b-field label="Changed by" horizontal>
<span>${transaction.user or ""}</span>
</b-field>
<b-field label="IP Address" horizontal>
<span>${transaction.remote_addr or ""}</span>
</b-field>
<b-field label="TXN ID" horizontal>
<span>${transaction.id}</span>
</b-field>
</div>
<div style="padding: 2rem;">
% for diff in version_diffs:
<h4 class="is-size-4 block">${diff.title} (${diff.operation_title})</h4>
${diff.render_html()}
% endfor
</div>
</%def>

20
src/wuttaweb/templates/master/view_versions.mako Normal file
View file

@ -0,0 +1,20 @@
## -*- coding: utf-8; -*-
<%inherit file="/page.mako" />
<%def name="title()">${index_title} » ${instance_title} » history</%def>
<%def name="content_title()">Version History</%def>
<%def name="page_content()">
${grid.render_vue_tag()}
</%def>
<%def name="render_vue_templates()">
${parent.render_vue_templates()}
${grid.render_vue_template()}
</%def>
<%def name="make_vue_components()">
${parent.make_vue_components()}
${grid.render_vue_finalize()}
</%def>

68
src/wuttaweb/testing.py
View file

@ -24,14 +24,18 @@
WuttaWeb - test utilities
"""
import sys
from unittest.mock import MagicMock
import fanstatic
import pytest
from pyramid import testing
from wuttjamaican.testing import DataTestCase
from wuttjamaican.db.model.base import metadata
from wuttaweb import subscribers
from wuttaweb.conf import WuttaWebConfigExtension
class WebTestCase(DataTestCase):
@ -101,4 +105,66 @@ class WebTestCase(DataTestCase):
"""
Make and return a new dummy request object.
"""
return testing.DummyRequest()
return testing.DummyRequest(client_addr="127.0.0.1")
@pytest.mark.versioned
class VersionWebTestCase(WebTestCase):
"""
Base class for test suites requiring a full (typical) web app,
with Continuum versioning support.
"""
def setUp(self):
self.setup_versioning()
def setup_versioning(self):
"""
Perform setup for the testing web app.
"""
self.setup_web()
def tearDown(self):
self.teardown_versioning()
def teardown_versioning(self):
"""
Perform teardown for the testing web app.
"""
import sqlalchemy_continuum as continuum # pylint: disable=import-outside-toplevel
continuum.remove_versioning()
continuum.versioning_manager.transaction_cls = continuum.TransactionFactory()
self.teardown_web()
def make_config(self, **kwargs):
"""
Make and customize the config object.
We override this to explicitly enable the versioning feature.
"""
from wutta_continuum.conf import ( # pylint: disable=import-outside-toplevel
WuttaContinuumConfigExtension,
)
config = super().make_config(**kwargs)
config.setdefault("wutta_continuum.enable_versioning", "true")
# nb. must purge model classes from sys.modules, so they will
# be reloaded and sqlalchemy-continuum can reconfigure
if "wuttjamaican.db.model" in sys.modules:
del sys.modules["wuttjamaican.db.model.batch"]
del sys.modules["wuttjamaican.db.model.upgrades"]
del sys.modules["wuttjamaican.db.model.auth"]
del sys.modules["wuttjamaican.db.model.base"]
del sys.modules["wuttjamaican.db.model"]
self.assertNotIn("user_version", metadata.tables)
ext = WuttaWebConfigExtension()
ext.configure(config)
ext = WuttaContinuumConfigExtension()
ext.startup(config)
return config

412
src/wuttaweb/views/master.py
View file

@ -25,6 +25,7 @@ Base Logic for Master Views
"""
# pylint: disable=too-many-lines
import datetime
import logging
import os
import threading
@ -33,13 +34,14 @@ import sqlalchemy as sa
from sqlalchemy import orm
from pyramid.renderers import render_to_response
from webhelpers2.html import HTML
from webhelpers2.html import HTML, tags
from wuttjamaican.util import get_class_hierarchy
from wuttaweb.views.base import View
from wuttaweb.util import get_form_data, render_csrf_token
from wuttaweb.db import Session
from wuttaweb.progress import SessionProgress
from wuttaweb.diffs import VersionDiff
log = logging.getLogger(__name__)
@ -348,6 +350,20 @@ class MasterView(View): # pylint: disable=too-many-public-methods
"configuring" - i.e. it should have a :meth:`configure()` view.
Default value is ``False``.
.. attribute:: has_versions
Boolean indicating whether the master view should expose
version history for its data records - i.e. it should have a
:meth:`view_versions()` view. Default value is ``False``.
See also :meth:`should_expose_versions()`.
.. attribute:: version_grid_columns
List of columns for the :meth:`view_versions()` view grid.
This is optional; see also :meth:`get_version_grid_columns()`.
**ROW FEATURES**
.. attribute:: has_rows
@ -435,6 +451,9 @@ class MasterView(View): # pylint: disable=too-many-public-methods
rows_paginate_on_backend = True
rows_viewable = False
# versioning features
has_versions = False
# current action
listing = False
creating = False
@ -891,6 +910,363 @@ class MasterView(View): # pylint: disable=too-many-public-methods
)
return form
##############################
# version history methods
##############################
@classmethod
def get_model_version_class(cls):
"""
Returns the version class for the master model class.
Should only be relevant if :attr:`has_versions` is true.
"""
import sqlalchemy_continuum as continuum # pylint: disable=import-outside-toplevel
return continuum.version_class(cls.get_model_class())
def should_expose_versions(self):
"""
Returns boolean indicating whether versioning history should
be exposed for the current user. This will return ``True``
unless any of the following are ``False``:
* :attr:`has_versions`
* :meth:`wuttjamaican:wuttjamaican.app.AppHandler.continuum_is_enabled()`
* ``self.has_perm("versions")`` - cf. :meth:`has_perm()`
:returns: ``True`` if versioning should be exposed for current
user; else ``False``.
"""
if not self.has_versions:
return False
if not self.app.continuum_is_enabled():
return False
if not self.has_perm("versions"):
return False
return True
def view_versions(self):
"""
View to list version history for an object. See also
:meth:`view_version()`.
This usually corresponds to a URL like
``/widgets/XXX/versions/`` where ``XXX`` represents the key/ID
for the record.
By default, this view is included only if :attr:`has_versions`
is true.
The default view logic will show a "grid" (table) with the
record's version history.
See also:
* :meth:`make_version_grid()`
"""
instance = self.get_instance()
instance_title = self.get_instance_title(instance)
grid = self.make_version_grid(instance)
# return grid data only, if partial page was requested
if self.request.GET.get("partial"):
context = grid.get_vue_context()
if grid.paginated and grid.paginate_on_backend:
context["pager_stats"] = grid.get_vue_pager_stats()
return self.json_response(context)
index_link = tags.link_to(self.get_index_title(), self.get_index_url())
instance_link = tags.link_to(
instance_title, self.get_action_url("view", instance)
)
index_title_rendered = HTML.literal("<span>&nbsp;&raquo;</span>").join(
[index_link, instance_link]
)
return self.render_to_response(
"view_versions",
{
"index_title_rendered": index_title_rendered,
"instance": instance,
"instance_title": instance_title,
"instance_url": self.get_action_url("view", instance),
"grid": grid,
},
)
def make_version_grid(self, instance=None, **kwargs):
"""
Create and return a grid for use with the
:meth:`view_versions()` view.
See also related methods, which are called by this one:
* :meth:`get_version_grid_key()`
* :meth:`get_version_grid_columns()`
* :meth:`get_version_grid_data()`
* :meth:`configure_version_grid()`
:returns: :class:`~wuttaweb.grids.base.Grid` instance
"""
import sqlalchemy_continuum as continuum # pylint: disable=import-outside-toplevel
route_prefix = self.get_route_prefix()
# instance = kwargs.pop("instance", None)
if not instance:
instance = self.get_instance()
if "key" not in kwargs:
kwargs["key"] = self.get_version_grid_key()
if "model_class" not in kwargs:
kwargs["model_class"] = continuum.transaction_class(self.get_model_class())
if "columns" not in kwargs:
kwargs["columns"] = self.get_version_grid_columns()
if "data" not in kwargs:
kwargs["data"] = self.get_version_grid_data(instance)
if "actions" not in kwargs:
route = f"{route_prefix}.version"
def url(txn, i): # pylint: disable=unused-argument
return self.request.route_url(route, uuid=instance.uuid, txnid=txn.id)
kwargs["actions"] = [
self.make_grid_action("view", icon="eye", url=url),
]
kwargs.setdefault("paginated", True)
grid = self.make_grid(**kwargs)
self.configure_version_grid(grid)
grid.load_settings()
return grid
@classmethod
def get_version_grid_key(cls):
"""
Returns the unique key to be used for the version grid, for caching
sort/filter options etc.
This is normally called automatically from :meth:`make_version_grid()`.
:returns: Grid key as string
"""
if hasattr(cls, "version_grid_key"):
return cls.version_grid_key
return f"{cls.get_route_prefix()}.history"
def get_version_grid_columns(self):
"""
Returns the default list of version grid column names, for the
:meth:`view_versions()` view.
This is normally called automatically by
:meth:`make_version_grid()`.
Subclass may define :attr:`version_grid_columns` for simple
cases, or can override this method if needed.
:returns: List of string column names
"""
if hasattr(self, "version_grid_columns"):
return self.version_grid_columns
return [
"id",
"issued_at",
"user",
"remote_addr",
]
def get_version_grid_data(self, instance):
"""
Returns the grid data query for the :meth:`view_versions()`
view.
This is normally called automatically by
:meth:`make_version_grid()`.
Default query will locate SQLAlchemy-Continuum ``transaction``
records which are associated with versions of the given model
instance. See also
:func:`wutta-continuum:wutta_continuum.util.model_transaction_query()`.
:returns: :class:`~sqlalchemy:sqlalchemy.orm.Query` instance
"""
import sqlalchemy_continuum as continuum # pylint: disable=import-outside-toplevel
from wutta_continuum.util import ( # pylint: disable=import-outside-toplevel
model_transaction_query,
)
model_class = self.get_model_class()
txncls = continuum.transaction_class(model_class)
query = model_transaction_query(instance)
return query.order_by(txncls.issued_at.desc())
def configure_version_grid(self, g):
"""
Configure the grid for the :meth:`view_versions()` view.
This is called automatically by :meth:`make_version_grid()`.
Default logic applies basic customization to the column labels etc.
"""
# id
g.set_label("id", "TXN ID")
# g.set_link("id")
# issued_at
g.set_label("issued_at", "Changed")
g.set_renderer("issued_at", self.render_issued_at)
g.set_link("issued_at")
g.set_sort_defaults("issued_at", "desc")
# user
g.set_label("user", "Changed by")
g.set_link("user")
# remote_addr
g.set_label("remote_addr", "IP Address")
def view_version(self): # pylint: disable=too-many-locals
"""
View to show diff details for a particular object version.
See also :meth:`view_versions()`.
This usually corresponds to a URL like
``/widgets/XXX/versions/YYY`` where ``XXX`` represents the
key/ID for the record and YYY represents a
SQLAlchemy-Continuum ``transaction.id``.
By default, this view is included only if :attr:`has_versions`
is true.
The default view logic will display a "diff" table showing how
the record's values were changed within a transaction.
See also:
* :func:`wutta-continuum:wutta_continuum.util.model_transaction_query()`
* :meth:`get_relevant_versions()`
* :class:`~wuttaweb.diffs.VersionDiff`
"""
import sqlalchemy_continuum as continuum # pylint: disable=import-outside-toplevel
from wutta_continuum.util import ( # pylint: disable=import-outside-toplevel
model_transaction_query,
)
instance = self.get_instance()
model_class = self.get_model_class()
route_prefix = self.get_route_prefix()
txncls = continuum.transaction_class(model_class)
transactions = model_transaction_query(instance)
txnid = self.request.matchdict["txnid"]
txn = transactions.filter(txncls.id == txnid).first()
if not txn:
raise self.notfound()
prev_url = None
older = (
transactions.filter(txncls.issued_at <= txn.issued_at)
.filter(txncls.id != txnid)
.order_by(txncls.issued_at.desc())
.first()
)
if older:
prev_url = self.request.route_url(
f"{route_prefix}.version", uuid=instance.uuid, txnid=older.id
)
next_url = None
newer = (
transactions.filter(txncls.issued_at >= txn.issued_at)
.filter(txncls.id != txnid)
.order_by(txncls.issued_at)
.first()
)
if newer:
next_url = self.request.route_url(
f"{route_prefix}.version", uuid=instance.uuid, txnid=newer.id
)
version_diffs = [
VersionDiff(version)
for version in self.get_relevant_versions(txn, instance)
]
index_link = tags.link_to(self.get_index_title(), self.get_index_url())
instance_title = self.get_instance_title(instance)
instance_link = tags.link_to(
instance_title, self.get_action_url("view", instance)
)
history_link = tags.link_to(
"history",
self.request.route_url(f"{route_prefix}.versions", uuid=instance.uuid),
)
index_title_rendered = HTML.literal("<span>&nbsp;&raquo;</span>").join(
[index_link, instance_link, history_link]
)
return self.render_to_response(
"view_version",
{
"index_title_rendered": index_title_rendered,
"instance": instance,
"instance_title": instance_title,
"instance_url": self.get_action_url("versions", instance),
"transaction": txn,
"changed": self.render_issued_at(txn, None, None),
"version_diffs": version_diffs,
"show_prev_next": True,
"prev_url": prev_url,
"next_url": next_url,
},
)
def get_relevant_versions(self, transaction, instance):
"""
Should return all version records pertaining to the given
model instance and transaction.
This is normally called from :meth:`view_version()`.
:param transaction: SQLAlchemy-Continuum ``transaction``
record/instance.
:param instance: Instance of the model class.
:returns: List of version records.
"""
session = self.Session()
vercls = self.get_model_version_class()
return (
session.query(vercls)
.filter(vercls.transaction == transaction)
.filter(vercls.uuid == instance.uuid)
.all()
)
def render_issued_at( # pylint: disable=missing-function-docstring,unused-argument
self, txn, key, value
):
dt = txn.issued_at
dt = dt.replace(tzinfo=datetime.timezone.utc)
dt = dt.astimezone(None)
return self.app.render_datetime(dt)
##############################
# autocomplete methods
##############################
@ -3065,7 +3441,10 @@ class MasterView(View): # pylint: disable=too-many-public-methods
cls._defaults(config)
@classmethod
def _defaults(cls, config):
def _defaults(cls, config): # pylint: disable=too-many-statements
wutta_config = config.registry.settings.get("wutta_config")
app = wutta_config.get_app()
route_prefix = cls.get_route_prefix()
permission_prefix = cls.get_permission_prefix()
url_prefix = cls.get_url_prefix()
@ -3171,6 +3550,7 @@ class MasterView(View): # pylint: disable=too-many-public-methods
# download
if cls.downloadable:
instance_url_prefix = cls.get_instance_url_prefix()
config.add_route(
f"{route_prefix}.download", f"{instance_url_prefix}/download"
)
@ -3188,6 +3568,7 @@ class MasterView(View): # pylint: disable=too-many-public-methods
# execute
if cls.executable:
instance_url_prefix = cls.get_instance_url_prefix()
config.add_route(
f"{route_prefix}.execute",
f"{instance_url_prefix}/execute",
@ -3235,3 +3616,30 @@ class MasterView(View): # pylint: disable=too-many-public-methods
config.add_wutta_permission(
permission_prefix, f"{permission_prefix}.view", f"View {model_title}"
)
# version history
if cls.has_versions and app.continuum_is_enabled():
instance_url_prefix = cls.get_instance_url_prefix()
config.add_wutta_permission(
permission_prefix,
f"{permission_prefix}.versions",
f"View version history for {model_title}",
)
config.add_route(
f"{route_prefix}.versions", f"{instance_url_prefix}/versions/"
)
config.add_view(
cls,
attr="view_versions",
route_name=f"{route_prefix}.versions",
permission=f"{permission_prefix}.versions",
)
config.add_route(
f"{route_prefix}.version", f"{instance_url_prefix}/versions/{{txnid}}"
)
config.add_view(
cls,
attr="view_version",
route_name=f"{route_prefix}.version",
permission=f"{permission_prefix}.versions",
)

3
tests/forms/test_schema.py
View file

@ -11,9 +11,10 @@ from pyramid import testing
from sqlalchemy import orm
from wuttjamaican.conf import WuttaConfig
from wuttjamaican.testing import DataTestCase
from wuttaweb.forms import schema as mod
from wuttaweb.forms import widgets
from wuttaweb.testing import DataTestCase, WebTestCase
from wuttaweb.testing import WebTestCase
class TestWuttaDateTime(TestCase):

4
tests/grids/test_filters.py
View file

@ -576,10 +576,6 @@ class TestDateAlchemyFilter(WebTestCase):
def setUp(self):
self.setup_web()
model = self.app.model
# nb. create table for TheLocalThing
model.Base.metadata.create_all(bind=self.session.bind)
self.sample_data = [
{"id": 1, "date": datetime.date(2024, 1, 1)},

222
tests/test_diffs.py Normal file
View file

@ -0,0 +1,222 @@
# -*- coding: utf-8; -*-
from wuttaweb import diffs as mod
from wuttaweb.testing import WebTestCase, VersionWebTestCase
# nb. using WebTestCase here only for mako support in render_html()
class TestDiff(WebTestCase):
def make_diff(self, *args, **kwargs):
return mod.Diff(*args, **kwargs)
def test_constructor(self):
old_data = {"foo": "bar"}
new_data = {"foo": "baz"}
diff = self.make_diff(old_data, new_data, fields=["foo"])
self.assertEqual(diff.fields, ["foo"])
def test_make_fields(self):
old_data = {"foo": "bar"}
new_data = {"foo": "bar", "baz": "zer"}
# nb. this calls make_fields()
diff = self.make_diff(old_data, new_data)
# TODO: should the fields be cumulative? or just use new_data?
self.assertEqual(diff.fields, ["baz", "foo"])
def test_values(self):
old_data = {"foo": "bar"}
new_data = {"foo": "baz"}
diff = self.make_diff(old_data, new_data)
self.assertEqual(diff.old_value("foo"), "bar")
self.assertEqual(diff.new_value("foo"), "baz")
def test_values_differ(self):
old_data = {"foo": "bar"}
new_data = {"foo": "baz"}
diff = self.make_diff(old_data, new_data)
self.assertTrue(diff.values_differ("foo"))
old_data = {"foo": "bar"}
new_data = {"foo": "bar"}
diff = self.make_diff(old_data, new_data)
self.assertFalse(diff.values_differ("foo"))
def test_render_values(self):
old_data = {"foo": "bar"}
new_data = {"foo": "baz"}
diff = self.make_diff(old_data, new_data)
self.assertEqual(diff.render_old_value("foo"), "'bar'")
self.assertEqual(diff.render_new_value("foo"), "'baz'")
def test_get_old_value_attrs(self):
# no change
old_data = {"foo": "bar"}
new_data = {"foo": "bar"}
diff = self.make_diff(old_data, new_data, nature="update")
self.assertEqual(diff.get_old_value_attrs(False), {})
# update
old_data = {"foo": "bar"}
new_data = {"foo": "baz"}
diff = self.make_diff(old_data, new_data, nature="update")
self.assertEqual(
diff.get_old_value_attrs(True),
{"style": f"background-color: {diff.old_color};"},
)
# delete
old_data = {"foo": "bar"}
new_data = {}
diff = self.make_diff(old_data, new_data, nature="delete")
self.assertEqual(
diff.get_old_value_attrs(True),
{"style": f"background-color: {diff.old_color};"},
)
def test_get_new_value_attrs(self):
# no change
old_data = {"foo": "bar"}
new_data = {"foo": "bar"}
diff = self.make_diff(old_data, new_data, nature="update")
self.assertEqual(diff.get_new_value_attrs(False), {})
# update
old_data = {"foo": "bar"}
new_data = {"foo": "baz"}
diff = self.make_diff(old_data, new_data, nature="update")
self.assertEqual(
diff.get_new_value_attrs(True),
{"style": f"background-color: {diff.new_color};"},
)
# create
old_data = {}
new_data = {"foo": "bar"}
diff = self.make_diff(old_data, new_data, nature="create")
self.assertEqual(
diff.get_new_value_attrs(True),
{"style": f"background-color: {diff.new_color};"},
)
def test_render_field_row(self):
old_data = {"foo": "bar"}
new_data = {"foo": "baz"}
diff = self.make_diff(old_data, new_data)
row = diff.render_field_row("foo")
self.assertIn("<tr>", row)
self.assertIn("&#39;bar&#39;", row)
self.assertIn(f'style="background-color: {diff.old_color};"', row)
self.assertIn("&#39;baz&#39;", row)
self.assertIn(f'style="background-color: {diff.new_color};"', row)
self.assertIn("</tr>", row)
def test_render_html(self):
old_data = {"foo": "bar"}
new_data = {"foo": "baz"}
diff = self.make_diff(old_data, new_data)
html = diff.render_html()
self.assertIn("<table", html)
self.assertIn("<tr>", html)
self.assertIn("&#39;bar&#39;", html)
self.assertIn(f'style="background-color: {diff.old_color};"', html)
self.assertIn("&#39;baz&#39;", html)
self.assertIn(f'style="background-color: {diff.new_color};"', html)
self.assertIn("</tr>", html)
self.assertIn("</table>", html)
class TestVersionDiff(VersionWebTestCase):
def make_diff(self, *args, **kwargs):
return mod.VersionDiff(*args, **kwargs)
def test_constructor(self):
import sqlalchemy_continuum as continuum
model = self.app.model
user = model.User(username="fred")
self.session.add(user)
self.session.commit()
user.username = "freddie"
self.session.commit()
self.session.delete(user)
self.session.commit()
txncls = continuum.transaction_class(model.User)
vercls = continuum.version_class(model.User)
versions = self.session.query(vercls).order_by(vercls.transaction_id).all()
self.assertEqual(len(versions), 3)
version = versions[0]
diff = self.make_diff(version)
self.assertEqual(diff.nature, "create")
self.assertEqual(
diff.fields,
["active", "password", "person_uuid", "prevent_edit", "username", "uuid"],
)
version = versions[1]
diff = self.make_diff(version)
self.assertEqual(diff.nature, "update")
self.assertEqual(
diff.fields,
["active", "password", "person_uuid", "prevent_edit", "username", "uuid"],
)
version = versions[2]
diff = self.make_diff(version)
self.assertEqual(diff.nature, "delete")
self.assertEqual(
diff.fields,
["active", "password", "person_uuid", "prevent_edit", "username", "uuid"],
)
def test_render_values(self):
import sqlalchemy_continuum as continuum
model = self.app.model
user = model.User(username="fred")
self.session.add(user)
self.session.commit()
user.username = "freddie"
self.session.commit()
self.session.delete(user)
self.session.commit()
txncls = continuum.transaction_class(model.User)
vercls = continuum.version_class(model.User)
versions = self.session.query(vercls).order_by(vercls.transaction_id).all()
self.assertEqual(len(versions), 3)
version = versions[0]
diff = self.make_diff(version)
self.assertEqual(diff.nature, "create")
self.assertEqual(diff.render_old_value("username"), "")
self.assertEqual(
diff.render_new_value("username"),
'<span style="font-family: monospace;">&#39;fred&#39;</span>',
)
version = versions[1]
diff = self.make_diff(version)
self.assertEqual(diff.nature, "update")
self.assertEqual(
diff.render_old_value("username"),
'<span style="font-family: monospace;">&#39;fred&#39;</span>',
)
self.assertEqual(
diff.render_new_value("username"),
'<span style="font-family: monospace;">&#39;freddie&#39;</span>',
)
version = versions[2]
diff = self.make_diff(version)
self.assertEqual(diff.nature, "delete")
self.assertEqual(
diff.render_old_value("username"),
'<span style="font-family: monospace;">&#39;freddie&#39;</span>',
)
self.assertEqual(diff.render_new_value("username"), "")

36
tests/util.py
View file

@ -1,43 +1,15 @@
# -*- coding: utf-8; -*-
from wuttjamaican.conf import WuttaConfig
from wuttjamaican.testing import FileConfigTestCase
from wuttaweb.menus import MenuHandler
class DataTestCase(FileConfigTestCase):
"""
Base class for test suites requiring a full (typical) database.
"""
def setUp(self):
self.setup_db()
def setup_db(self):
self.setup_files()
self.config = WuttaConfig(
defaults={
"wutta.db.default.url": "sqlite://",
}
)
self.app = self.config.get_app()
# init db
model = self.app.model
model.Base.metadata.create_all(bind=self.config.appdb_engine)
self.session = self.app.make_session()
def tearDown(self):
self.teardown_db()
def teardown_db(self):
self.teardown_files()
class NullMenuHandler(MenuHandler):
"""
Dummy menu handler for testing.
Dummy :term:`menu handler` for testing.
"""
def make_menus(self, request, **kwargs):
"""
This always returns an empty menu set.
"""
return []

32
tests/views/test_batch.py
View file

@ -31,12 +31,6 @@ class MockBatchHandler(BatchHandler):
class TestBatchMasterView(WebTestCase):
def setUp(self):
self.setup_web()
# nb. create MockBatch, MockBatchRow
model.Base.metadata.create_all(bind=self.session.bind)
def make_handler(self):
return MockBatchHandler(self.config)
@ -51,7 +45,7 @@ class TestBatchMasterView(WebTestCase):
self.assertEqual(view.batch_handler, 42)
def test_get_fallback_templates(self):
handler = MockBatchHandler(self.config)
handler = self.make_handler()
with patch.object(
mod.BatchMasterView, "get_batch_handler", return_value=handler
):
@ -67,7 +61,7 @@ class TestBatchMasterView(WebTestCase):
def test_render_to_response(self):
model = self.app.model
handler = MockBatchHandler(self.config)
handler = self.make_handler()
user = model.User(username="barney")
self.session.add(user)
@ -87,7 +81,7 @@ class TestBatchMasterView(WebTestCase):
self.assertIs(context["batch_handler"], handler)
def test_configure_grid(self):
handler = MockBatchHandler(self.config)
handler = self.make_handler()
with patch.multiple(mod.BatchMasterView, create=True, model_class=MockBatch):
with patch.object(
mod.BatchMasterView, "get_batch_handler", return_value=handler
@ -98,7 +92,7 @@ class TestBatchMasterView(WebTestCase):
view.configure_grid(grid)
def test_render_batch_id(self):
handler = MockBatchHandler(self.config)
handler = self.make_handler()
with patch.object(
mod.BatchMasterView, "get_batch_handler", return_value=handler
):
@ -112,7 +106,7 @@ class TestBatchMasterView(WebTestCase):
self.assertIsNone(result)
def test_get_instance_title(self):
handler = MockBatchHandler(self.config)
handler = self.make_handler()
with patch.object(
mod.BatchMasterView, "get_batch_handler", return_value=handler
):
@ -127,7 +121,7 @@ class TestBatchMasterView(WebTestCase):
self.assertEqual(result, "00000043 runnin some numbers")
def test_configure_form(self):
handler = MockBatchHandler(self.config)
handler = self.make_handler()
with patch.multiple(mod.BatchMasterView, create=True, model_class=MockBatch):
with patch.object(
mod.BatchMasterView, "get_batch_handler", return_value=handler
@ -163,7 +157,7 @@ class TestBatchMasterView(WebTestCase):
view.configure_form(form)
def test_objectify(self):
handler = MockBatchHandler(self.config)
handler = self.make_handler()
with patch.multiple(mod.BatchMasterView, create=True, model_class=MockBatch):
with patch.object(
mod.BatchMasterView, "get_batch_handler", return_value=handler
@ -194,7 +188,7 @@ class TestBatchMasterView(WebTestCase):
def test_redirect_after_create(self):
self.pyramid_config.add_route("mock_batches.view", "/batch/mock/{uuid}")
handler = MockBatchHandler(self.config)
handler = self.make_handler()
with patch.object(
mod.BatchMasterView, "get_batch_handler", return_value=handler
):
@ -247,7 +241,7 @@ class TestBatchMasterView(WebTestCase):
def test_populate_thread(self):
model = self.app.model
handler = MockBatchHandler(self.config)
handler = self.make_handler()
with patch.object(
mod.BatchMasterView, "get_batch_handler", return_value=handler
):
@ -315,7 +309,7 @@ class TestBatchMasterView(WebTestCase):
def test_execute(self):
self.pyramid_config.add_route("mock_batches.view", "/batch/mock/{uuid}")
model = self.app.model
handler = MockBatchHandler(self.config)
handler = self.make_handler()
user = model.User(username="barney")
self.session.add(user)
@ -345,7 +339,7 @@ class TestBatchMasterView(WebTestCase):
self.assertTrue(self.request.session.peek_flash("error"))
def test_get_row_model_class(self):
handler = MockBatchHandler(self.config)
handler = self.make_handler()
with patch.object(
mod.BatchMasterView, "get_batch_handler", return_value=handler
):
@ -370,7 +364,7 @@ class TestBatchMasterView(WebTestCase):
self.assertIs(cls, MockBatchRow)
def test_get_row_grid_data(self):
handler = MockBatchHandler(self.config)
handler = self.make_handler()
model = self.app.model
user = model.User(username="barney")
@ -401,7 +395,7 @@ class TestBatchMasterView(WebTestCase):
self.assertEqual(data.count(), 1)
def test_configure_row_grid(self):
handler = MockBatchHandler(self.config)
handler = self.make_handler()
model = self.app.model
user = model.User(username="barney")

247
tests/views/test_master.py
View file

@ -16,7 +16,8 @@ from wuttaweb.views import master as mod
from wuttaweb.views import View
from wuttaweb.progress import SessionProgress
from wuttaweb.subscribers import new_request_set_user
from wuttaweb.testing import WebTestCase
from wuttaweb.testing import WebTestCase, VersionWebTestCase
from wuttaweb.grids import Grid
class TestMasterView(WebTestCase):
@ -1841,3 +1842,247 @@ class TestMasterView(WebTestCase):
# class may specify
with patch.object(view, "rows_title", create=True, new="Mock Rows"):
self.assertEqual(view.get_rows_title(), "Mock Rows")
class TestVersionedMasterView(VersionWebTestCase):
def make_view(self):
return mod.MasterView(self.request)
def test_defaults(self):
model = self.app.model
with patch.multiple(mod.MasterView, model_class=model.User, has_versions=True):
mod.MasterView.defaults(self.pyramid_config)
def test_get_model_version_class(self):
model = self.app.model
with patch.object(mod.MasterView, "model_class", new=model.User):
view = self.make_view()
vercls = view.get_model_version_class()
self.assertEqual(vercls.__name__, "UserVersion")
def test_should_expose_versions(self):
model = self.app.model
with patch.multiple(mod.MasterView, model_class=model.User, has_versions=True):
# fully enabled for root user
with patch.object(self.request, "is_root", new=True):
view = self.make_view()
self.assertTrue(view.should_expose_versions())
# but not if user has no access
view = self.make_view()
self.assertFalse(view.should_expose_versions())
# again, works for root user
with patch.object(self.request, "is_root", new=True):
view = self.make_view()
self.assertTrue(view.should_expose_versions())
# but not if config disables versioning
with patch.object(view.app, "continuum_is_enabled", return_value=False):
self.assertFalse(view.should_expose_versions())
def test_get_version_grid_key(self):
model = self.app.model
with patch.object(mod.MasterView, "model_class", new=model.User):
# default
view = self.make_view()
self.assertEqual(view.get_version_grid_key(), "users.history")
# custom
with patch.object(
mod.MasterView,
"version_grid_key",
new="users_custom_history",
create=True,
):
view = self.make_view()
self.assertEqual(view.get_version_grid_key(), "users_custom_history")
def test_get_version_grid_columns(self):
model = self.app.model
with patch.object(mod.MasterView, "model_class", new=model.User):
# default
view = self.make_view()
self.assertEqual(
view.get_version_grid_columns(),
["id", "issued_at", "user", "remote_addr"],
)
# custom
with patch.object(
mod.MasterView,
"version_grid_columns",
new=["issued_at", "user"],
create=True,
):
view = self.make_view()
self.assertEqual(view.get_version_grid_columns(), ["issued_at", "user"])
def test_get_version_grid_data(self):
model = self.app.model
user = model.User(username="fred")
self.session.add(user)
self.session.commit()
user.username = "freddie"
self.session.commit()
with patch.object(mod.MasterView, "model_class", new=model.User):
view = self.make_view()
query = view.get_version_grid_data(user)
self.assertIsInstance(query, orm.Query)
transactions = query.all()
self.assertEqual(len(transactions), 2)
def test_configure_version_grid(self):
import sqlalchemy_continuum as continuum
model = self.app.model
txncls = continuum.transaction_class(model.User)
with patch.object(mod.MasterView, "model_class", new=model.User):
view = self.make_view()
# this is mostly just for coverage, but we at least can
# confirm something does change
grid = view.make_grid(model_class=txncls)
self.assertNotIn("issued_at", grid.linked_columns)
view.configure_version_grid(grid)
self.assertIn("issued_at", grid.linked_columns)
def test_make_version_grid(self):
model = self.app.model
user = model.User(username="fred")
self.session.add(user)
self.session.commit()
user.username = "freddie"
self.session.commit()
with patch.object(mod.MasterView, "model_class", new=model.User):
with patch.object(mod.MasterView, "Session", return_value=self.session):
with patch.dict(self.request.matchdict, uuid=user.uuid):
view = self.make_view()
grid = view.make_version_grid()
self.assertIsInstance(grid, Grid)
self.assertIsInstance(grid.data, orm.Query)
self.assertEqual(len(grid.data.all()), 2)
def test_view_versions(self):
self.pyramid_config.add_route("home", "/")
self.pyramid_config.add_route("login", "/auth/login")
self.pyramid_config.add_route("users", "/users/")
self.pyramid_config.add_route("users.view", "/users/{uuid}")
self.pyramid_config.add_route("users.version", "/users/{uuid}/versions/{txnid}")
model = self.app.model
user = model.User(username="fred")
self.session.add(user)
self.session.commit()
user.username = "freddie"
self.session.commit()
with patch.object(mod.MasterView, "model_class", new=model.User):
with patch.object(mod.MasterView, "Session", return_value=self.session):
with patch.dict(self.request.matchdict, uuid=user.uuid):
view = self.make_view()
# normal, full page
response = view.view_versions()
self.assertEqual(response.content_type, "text/html")
self.assertIn("<b-table", response.text)
# partial page
with patch.dict(self.request.params, partial="1"):
response = view.view_versions()
self.assertEqual(response.content_type, "application/json")
self.assertIn("data", response.json)
self.assertEqual(len(response.json["data"]), 2)
def test_get_relevant_versions(self):
import sqlalchemy_continuum as continuum
model = self.app.model
txncls = continuum.transaction_class(model.User)
vercls = continuum.version_class(model.User)
user = model.User(username="fred")
self.session.add(user)
self.session.commit()
txn = (
self.session.query(txncls)
.join(vercls, vercls.transaction_id == txncls.id)
.order_by(txncls.id)
.first()
)
with patch.object(mod.MasterView, "model_class", new=model.User):
with patch.object(mod.MasterView, "Session", return_value=self.session):
view = self.make_view()
versions = view.get_relevant_versions(txn, user)
self.assertEqual(len(versions), 1)
version = versions[0]
self.assertIsInstance(version, vercls)
def test_view_version(self):
import sqlalchemy_continuum as continuum
self.pyramid_config.add_route("home", "/")
self.pyramid_config.add_route("login", "/auth/login")
self.pyramid_config.add_route("users", "/users/")
self.pyramid_config.add_route("users.view", "/users/{uuid}")
self.pyramid_config.add_route("users.versions", "/users/{uuid}/versions/")
self.pyramid_config.add_route("users.version", "/users/{uuid}/versions/{txnid}")
model = self.app.model
txncls = continuum.transaction_class(model.User)
vercls = continuum.version_class(model.User)
user = model.User(username="fred")
self.session.add(user)
self.session.commit()
user.username = "freddie"
self.session.commit()
transactions = (
self.session.query(txncls)
.join(vercls, vercls.transaction_id == txncls.id)
.order_by(txncls.id)
.all()
)
self.assertEqual(len(transactions), 2)
with patch.object(mod.MasterView, "model_class", new=model.User):
with patch.object(mod.MasterView, "Session", return_value=self.session):
# invalid txnid
with patch.dict(self.request.matchdict, uuid=user.uuid, txnid=999999):
view = self.make_view()
self.assertRaises(HTTPNotFound, view.view_version)
# first txn
first = transactions[0]
with patch.dict(self.request.matchdict, uuid=user.uuid, txnid=first.id):
view = self.make_view()
response = view.view_version()
self.assertIn(
'<table class="table is-fullwidth is-bordered is-narrow">',
response.text,
)
# second txn
second = transactions[1]
with patch.dict(
self.request.matchdict, uuid=user.uuid, txnid=second.id
):
view = self.make_view()
response = view.view_version()
self.assertIn(
'<table class="table is-fullwidth is-bordered is-narrow">',
response.text,
)

9
tox.ini
View file

@ -4,10 +4,13 @@ envlist = py38, py39, py310, py311, nox
[testenv]
extras = continuum,tests
commands = pytest {posargs}
commands =
pytest -m 'not versioned' {posargs}
pytest -m 'versioned' {posargs}
[testenv:nox]
extras = tests
commands = pytest -m 'not versioned' {posargs}
[testenv:pylint]
basepython = python3.11
@ -15,7 +18,9 @@ commands = pylint wuttaweb
[testenv:coverage]
basepython = python3.11
commands = pytest --cov=wuttaweb --cov-report=html --cov-fail-under=100
commands =
pytest -m 'not versioned' --cov=wuttaweb
pytest -m 'versioned' --cov-append --cov=wuttaweb --cov-report=html --cov-fail-under=100
[testenv:docs]
basepython = python3.11