apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-05-02 11:52:37 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Fix flow loading (#611)

Browse source
This commit is contained in:
cybermaggedon 2026-01-14 16:23:15 +00:00 • committed by GitHub
parent 387afee7b7
commit 62b754d788
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 15 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

8
tests/unit/test_gateway/test_config_receiver.py
View file

@ -64,7 +64,7 @@ class TestConfigReceiver:
mock_msg.value.return_value = Mock(
version="1.0",
config={
"flows": {
"flow": {
"flow1": '{"name": "test_flow_1", "steps": []}',
"flow2": '{"name": "test_flow_2", "steps": []}'
}
@ -109,7 +109,7 @@ class TestConfigReceiver:
mock_msg.value.return_value = Mock(
version="1.0",
config={
"flows": {
"flow": {
"flow1": '{"name": "test_flow_1", "steps": []}'
}
}
@ -352,7 +352,7 @@ class TestConfigReceiver:
mock_msg.value.return_value = Mock(
version="1.0",
config={
"flows": {
"flow": {
"flow2": '{"name": "test_flow_2", "steps": []}',
"flow3": '{"name": "test_flow_3", "steps": []}'
}
@ -393,7 +393,7 @@ class TestConfigReceiver:
mock_msg.value.return_value = Mock(
version="1.0",
config={
"flows": {
"flow": {
"flow1": '{"invalid": json}', # Invalid JSON
"flow2": '{"name": "valid_flow", "steps": []}' # Valid JSON
}

24
trustgraph-flow/trustgraph/gateway/config/receiver.py
View file

@ -53,22 +53,20 @@ class ConfigReceiver:
logger.info(f"Config version: {v.version}")
if "flows" in v.config:
flows = v.config.get("flow", {})
flows = v.config["flows"]
wanted = list(flows.keys())
current = list(self.flows.keys())
wanted = list(flows.keys())
current = list(self.flows.keys())
for k in wanted:
if k not in current:
self.flows[k] = json.loads(flows[k])
await self.start_flow(k, self.flows[k])
for k in wanted:
if k not in current:
self.flows[k] = json.loads(flows[k])
await self.start_flow(k, self.flows[k])
for k in current:
if k not in wanted:
await self.stop_flow(k, self.flows[k])
del self.flows[k]
for k in current:
if k not in wanted:
await self.stop_flow(k, self.flows[k])
del self.flows[k]
except Exception as e:
logger.error(f"Config processing exception: {e}", exc_info=True)