Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: set entity timestamp (at source) if not provided #36

Merged
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/CODEOWNERS
Original file line number Diff line number Diff line change
@@ -1 +1 @@
* @leoparente @mfiedorowicz
* @leoparente @ltucker @mfiedorowicz
5 changes: 5 additions & 0 deletions netboxlabs/diode/sdk/ingester.py
Original file line number Diff line number Diff line change
Expand Up @@ -710,6 +710,11 @@ def __new__(
virtual_machine, VirtualMachinePb, name=virtual_machine
)

if timestamp is None:
ts = _timestamp_pb2.Timestamp()
ts.GetCurrentTime()
timestamp = ts

return EntityPb(
site=site,
platform=platform,
Expand Down
33 changes: 33 additions & 0 deletions tests/test_ingester.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
#!/usr/bin/env python
# Copyright 2024 NetBox Labs Inc
"""NetBox Labs - Tests."""
from google.protobuf import timestamp_pb2

# ruff: noqa: I001
from netboxlabs.diode.sdk.diode.v1.ingester_pb2 import (
Expand Down Expand Up @@ -658,6 +659,38 @@ def test_site_instantiation_with_all_fields():
assert isinstance(tag, TagPb)


def test_entity_instantiation_with_no_timestamp_provided():
"""Check Entity instantiation with no timestamp provided."""
entity = Entity(
site="Site1",
)
assert isinstance(entity, EntityPb)
assert isinstance(entity.site, SitePb)
assert entity.site.name == "Site1"
assert entity.timestamp is not None
assert entity.timestamp.seconds > 0
assert entity.timestamp.nanos > 0


def test_entity_instantiation_with_timestamp_provided():
"""Check Entity instantiation with timestamp provided."""
current_ts = timestamp_pb2.Timestamp()
current_ts.GetCurrentTime()

entity = Entity(
site="Site1",
timestamp=current_ts,
)
assert isinstance(entity, EntityPb)
assert isinstance(entity.site, SitePb)
assert entity.site.name == "Site1"
assert entity.timestamp is not None
assert entity.timestamp.seconds > 0
assert entity.timestamp.nanos > 0
assert entity.timestamp.seconds == current_ts.seconds
assert entity.timestamp.nanos == current_ts.nanos


def test_entity_instantiation_with_site():
"""Check Entity instantiation with site."""
entity = Entity(
Expand Down
Loading