feat: introduce TopicFilter support

[matthax]

Hey, I'll start by saying I'm new to this and I just ran into an issue where I could not control topic subscriptions, so I've tried to add it a few ways and I wanted to bring that to this PR and get some feedback.

My first attempt was to simply set the Filter something like

subscription_params["Filter"] = {
    "_value_1": "*",
}

This doesn't break, but it doesn't actually work. The actual definition can be found in the spec

<xsd:complexType name="TopicExpressionType" mixed="true">
  <xsd:simpleContent>
    <xsd:extension base="xsd:string">
      <xsd:attribute name="Dialect" type="xsd:anyURI" use="required" />
      <xsd:anyAttribute/>
    </xsd:extension>
  </xsd:simpleContent>
</xsd:complexType>

So I added a constructor to types.py but I'm not sure that's really right, and the reason is because I'm relying on the zeep client to actually build the TopicExpressionType. Maybe there's a better way to do this without requiring the client binding?

But, this approach does make it easy for clients to call in, you just specify a filter as a string, and the dialect is set automatically for you. Painless, but you lose the ability to specify a different dialect (This could also be added as a parameter to the constructor too, if you'd prefer that). It also makes it a little easier to do some validation, (e.g. the XPATH is valid syntactically).

Another approach would be to allow the filter to be passed in fully, and make it the callers problem to build the TopicExpression. Maybe it's still worthwhile to provide a constructor or utility to make life easier, but that would give callers the ability to set their own dialect and move the whole zeep client dependency coupling totally out of the manager. It's also worth noting that cameras with FixedTopicSet = True don't seem to actually support XPATHs, they expect a specific selection. If the selection is invalid, you'll get a pretty gnarly exception. In my application I use GetEventProperties to validate the selection, it didn't seem to be something that would really fit in the manager class, but if you wanted that as an example or something I could probably pull it together.

Let me know what you think would be the right approach or if I can make any changes, just thought it was useful to limit the subscriptions.

I also fixed an issue with the MINIMUM_SUBSCRIPTION_SECONDS, I should probably make that a different PR but if you specify an interval under the 60s, the subscription won't renew in time, subsequent calls fail on connection timeouts and things are sad. I just enforced the minimum to get around this.

[codecov]

Codecov Report

❌ Patch coverage is 56.00000% with 11 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
onvif/managers.py	36.36%	7 Missing ⚠️
onvif/types.py	76.92%	3 Missing ⚠️
onvif/client.py	0.00%	1 Missing ⚠️

Files with missing lines	Coverage Δ
onvif/client.py	67.28% <0.00%> (ø)
onvif/types.py	91.02% <76.92%> (-2.83%)	⬇️
onvif/managers.py	32.72% <36.36%> (+0.67%)	⬆️

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[jamesst20]

Thank you @matthax you've helped me figure out why I couldn't get Subscribe to work.

I kept getting errors like

Any element received object of type 'str', expected lxml.etree._Element or builtins.dict or zeep.xsd.valueobjects.AnyObject
See http://docs.python-zeep.org/en/master/datastructures.html#any-objects for more information

and I ended up on this PR and I also had the same error with your implementation. This is when I figured out something else must have been wrong instead of my payload and I ended up editing b-2.xsd like this:

and it worked right away! Just though I would share! In my case I went for a simpler approach that is equivalent to yours I believe

        if filter_expression:
            topic_expr_element = self.client.get_element("wsnt:TopicExpression")
            topic_expr_type = self.client.get_type("wsnt:TopicExpressionType")
            topic_expr = topic_expr_type(
                filter_expression, Dialect="http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet"
            )
            any_topic_expr = AnyObject(topic_expr_element, topic_expr)
            message["Filter"] = {"_value_1": [any_topic_expr]}

I am not sure why the definitions of b-2.xsd are different for TopicExpressionType here http://docs.oasis-open.org/wsn/b-2.xsd

[matthax]

Thank you @matthax you've helped me figure out why I couldn't get Subscribe to work.

I kept getting errors like

Any element received object of type 'str', expected lxml.etree._Element or builtins.dict or zeep.xsd.valueobjects.AnyObject

See http://docs.python-zeep.org/en/master/datastructures.html#any-objects for more information

and I ended up on this PR and I also had the same error with your implementation. This is when I figured out something else must have been wrong instead of my payload and I ended up editing b-2.xsd like this:

and it worked right away! Just though I would share! In my case I went for a simpler approach that is equivalent to yours I believe

        if filter_expression:

            topic_expr_element = self.client.get_element("wsnt:TopicExpression")

            topic_expr_type = self.client.get_type("wsnt:TopicExpressionType")

            topic_expr = topic_expr_type(

                filter_expression, Dialect="http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet"

            )

            any_topic_expr = AnyObject(topic_expr_element, topic_expr)

            message["Filter"] = {"_value_1": [any_topic_expr]}

I am not sure why the definitions of b-2.xsd are different for TopicExpressionType here http://docs.oasis-open.org/wsn/b-2.xsd

Ah! I should have thought of just editing the file, this is much cleaner as an approach, nice work! Maybe we just need that change in the upstream onvif library then and I could clean up the manual construction I'm doing?

[jamesst20]

<xsd:extension base="xsd:string">

You don't need to edit the file, I've actually extracted the definition from https://github.com/openvideolibs/python-onvif-zeep-async/blob/async/onvif/wsdl/b-2.xsd#L51 because the latest version here http://docs.oasis-open.org/wsn/b-2.xsd appears to not be working. My demo should work for python-onvif-zeep-async as well without any extra work.

In case it complains about not finding the namespace wsnt, you can simply register the alias to the client or use the full URL

client.set_ns_prefix("wsnt", "http://docs.oasis-open.org/wsn/b-2")

[bdraco]

codecov/patchFailing after 1s — 56.00% of diff hit (target 68.83%)

Please complete test coverage

[bdraco]

@bluetoothbot review

[bluetoothbot]

@matthax: > Thank you @matthax you've helped me figure out why I couldn't get Subscribe to work.

On cleaning up the manual construction: you actually don't need any upstream/xsd change for this repo. The bundled onvif/wsdl/b-2.xsd already defines TopicExpressionType as <xsd:extension base="xsd:string"> (lines 49-56) — that's the patched definition @jamesst20 was pasting, so client.get_type("{http://docs.oasis-open.org/wsn/b-2}TopicExpressionType") resolves correctly out of the box here.

The AnyObject wrapping can't be fully eliminated, though, and that's by design rather than a workaround. Filter is a FilterType whose body is <xsd:any minOccurs="0" maxOccurs="unbounded"/> (b-2.xsd lines 58-62). For an xsd:any, zeep can't infer the element/type from the schema, so it requires you to hand it an AnyObject (element + value) to serialize — passing a raw str is exactly what triggers the "Any element received object of type 'str'..." error. So you can simplify the construction (which is what your TopicExpression/from_client helper does nicely) but the Element + value pairing has to stay.

Two concrete suggestions while you're here: because that xsd:any is maxOccurs="unbounded", prefer the list form {"_value_1": [topic_expr]} (as in @jamesst20's snippet) over a single object — it matches the schema cardinality and is the known-working shape. And if a device complains about the wsnt prefix, client.set_ns_prefix("wsnt", "http://docs.oasis-open.org/wsn/b-2") is the fix, though using the full {...} namespace in get_type (as your helper does) sidesteps that entirely.

[bluetoothbot]

PR Review — feat: introduce TopicFilter support

The topic-filter feature is a useful addition and the TopicExpression/from_client approach is reasonable — imports resolve, events_service.zeep_client is valid, and the bundled b-2.xsd already carries the patched TopicExpressionType so no schema edit is needed. However, there are two blocking issues. (1) The MINIMUM_SUBSCRIPTION_INTERVAL clamp in BaseManager.__init__ is inverted: interval if interval <= MIN else MIN keeps sub-60s intervals and caps everything above 60s down to 60s — the opposite of enforcing a floor, and a regression for existing callers across both managers. Use max(interval, MINIMUM_SUBSCRIPTION_INTERVAL). (2) Test coverage is failing per codecov and @bdraco's change request; none of the new code paths are exercised. Secondary issues: the topic_filter annotation in the manager (TopicExpression | None) contradicts client.py and the actual string usage; the filter is validated as XPath but always sent with the ConcreteSet dialect with no override; and the Filter._value_1 value should likely be a list per the schema's unbounded xsd:any. I'd also split the interval fix into its own PR as you suggested. Address the inverted clamp and add tests, and this is close.

---

🔴 Blocking

1. Minimum-interval clamp is inverted — caps the interval instead of enforcing a floor (`onvif/managers.py`, L58-62)

This logic is backwards and does the opposite of what the PR intends. MINIMUM_SUBSCRIPTION_SECONDS is a floor ("if the camera returns a termination time less than this value, we use this value instead"), so sub-60s intervals should be raised to 60s and larger intervals left untouched.

Walk through the current expression:

• interval = 30s → 30 <= 60 is True → keeps 30s (the too-small value the PR set out to prevent).
• interval = 600s → 600 <= 60 is False → collapses to 60s.

So it fails to enforce the minimum and it caps every interval above 60s down to 60s. That second effect is a regression for all existing callers: downstream consumers (e.g. Home Assistant's ONVIF integration) pass intervals far larger than a minute, and this change would force them to renew roughly every ~48s (60s × _RENEWAL_PERCENTAGE 0.8), dramatically increasing subscription/renewal traffic.

Use a floor instead:

self._interval = max(interval, MINIMUM_SUBSCRIPTION_INTERVAL)

Note this mirrors the existing, correct pattern already in _calculate_next_renewal_call_at at line 126: max(delay.total_seconds(), MINIMUM_SUBSCRIPTION_SECONDS).

Also note this clamp lives in BaseManager.__init__, so it affects NotificationManager (push) too, not just PullPoint — which is fine with max(), but makes the inverted behavior more damaging. As the PR description itself acknowledges, this fix is unrelated to the topic-filter feature and would be cleaner as its own PR.

self._interval = (
    interval
    if interval <= MINIMUM_SUBSCRIPTION_INTERVAL
    else MINIMUM_SUBSCRIPTION_INTERVAL
)

2. New TopicExpression / filter paths are untested (blocking per codecov + review) (`onvif/types.py`, L107-127)

@bdraco requested changes and codecov is failing (56% of diff covered vs 68.83% target). None of the new logic is exercised:

• TopicExpression.__init__ and TopicExpression.from_client (this file)
• PullPointManager.__init__ filter validation (the XPath(...).path branch and the else None branch)
• the subscription_params["Filter"] branch in PullPointManager._start
• the new interval-clamp branch in BaseManager.__init__

tests/test_types.py already constructs an ONVIFCamera against the bundled WSDL and parses SOAP envelopes, so it's a good home for TopicExpression tests. Suggested coverage:

1. TopicExpression.from_client(client, "tns1:.../Motion") produces an AnyObject whose serialized XML carries the expected Dialect attribute and _value_1 text.
2. A PullPointManager(...) built with a valid filter stores the normalized path; with None stores None.
3. An invalid filter (e.g. "[") raises XPathSyntaxError from the constructor.
4. The interval clamp: a 30s interval yields a 60s effective interval, and a 600s interval is preserved (this test will currently fail and catch the inverted-logic bug above).

🟡 Important

1. topic_filter annotation contradicts client.py and the actual usage (`onvif/managers.py`, L306)

The parameter is annotated topic_filter: TopicExpression | None, but everything else treats it as a plain XPath string:

• client.py::create_pullpoint_manager declares it topic_filter: str | None = None (line 562-563) and forwards it directly.
• the body does XPath(topic_filter).path and stores self._topic_filter: str | None.
• the docstring describes it as "An optional XPATH" and the example passes the raw string "tns1:RuleEngine/CellMotionDetector/Motion".

A caller passing an actual TopicExpression object here would break, since lxml.etree.XPath() expects a string. Change the annotation to topic_filter: str | None = None so the manager and the client agree.

topic_filter: TopicExpression | None = None,

2. Filter validated as XPath but always sent with the ConcreteSet dialect (`onvif/managers.py`, L325-328)

XPath(topic_filter).path validates the expression against lxml's XPath grammar, but the value is always serialized with DIALECT = ".../topicExpression/ConcreteSet" (the hardcoded default in TopicExpression.from_client). These are different ONVIF topic dialects: ConcreteSet expressions and XPath expressions don't have identical grammars, so this can both reject valid ConcreteSet filters and accept strings that the camera will reject. As you noted in the description, FixedTopicSet=True cameras don't accept XPath at all.

Two follow-on points:

1. There's no way to override the dialect through the public API — create_pullpoint_manager/PullPointManager always go through from_client with the default. If you want to support XPath vs ConcreteSet, plumb a dialect parameter through and only run the lxml XPath check when the dialect is actually XPath.
2. XPath(topic_filter).path is effectively a parse-and-return-original round-trip; the only effect is the syntax check via the raised XPathSyntaxError. That's fine, but worth a brief comment so a future reader doesn't think it normalizes the expression.

self._topic_filter: str | None = (
    XPath(topic_filter).path if topic_filter else None
)

🟢 Suggestions

1. Filter `_value_1` may need to be a list (`onvif/managers.py`, L349-352)

FilterType in the bundled onvif/wsdl/b-2.xsd (lines 58-62) defines its body as <xsd:any minOccurs="0" maxOccurs="unbounded"/>. zeep typically represents an unbounded any as a list, and @jamesst20's working snippet in the comments wraps it accordingly: message["Filter"] = {"_value_1": [any_topic_expr]}. You're assigning a single AnyObject. It may serialize correctly with some zeep versions, but wrapping it in a list matches the schema cardinality and the known-working example — worth confirming against a real device and using the list form for robustness.

subscription_params["Filter"] = {
    "_value_1": TopicExpression.from_client(
        events_service.zeep_client, self._topic_filter
    ),
}

---

Checklist

• No hardcoded secrets
• Minimum subscription interval correctly enforced — critical ＃1
• Test coverage for new code paths — critical ＃2
• Type annotations consistent across call sites — warning ＃1
• Input validation matches the wire format/dialect — warning ＃2
• No mutable default arguments
• Imports resolve and attributes exist

---

Summary

The topic-filter feature is a useful addition and the TopicExpression/from_client approach is reasonable — imports resolve, events_service.zeep_client is valid, and the bundled b-2.xsd already carries the patched TopicExpressionType so no schema edit is needed. However, there are two blocking issues. (1) The MINIMUM_SUBSCRIPTION_INTERVAL clamp in BaseManager.__init__ is inverted: interval if interval <= MIN else MIN keeps sub-60s intervals and caps everything above 60s down to 60s — the opposite of enforcing a floor, and a regression for existing callers across both managers. Use max(interval, MINIMUM_SUBSCRIPTION_INTERVAL). (2) Test coverage is failing per codecov and @bdraco's change request; none of the new code paths are exercised. Secondary issues: the topic_filter annotation in the manager (TopicExpression | None) contradicts client.py and the actual string usage; the filter is validated as XPath but always sent with the ConcreteSet dialect with no override; and the Filter._value_1 value should likely be a list per the schema's unbounded xsd:any. I'd also split the interval fix into its own PR as you suggested. Address the inverted clamp and add tests, and this is close.

---

To rebase specific severity levels, mention me: @bluetoothbot rebase critical (fixes 🔴 only), @bluetoothbot rebase important (fixes 🔴 + 🟡), or just @bluetoothbot rebase for all.

---

Automated review by Kōan340e9ea
5c5aad3
2893a07
c8e6015
8ad2a72