Add distributor api projectdev-mod

Issue-ID: DCAEGEN2-1860
Signed-off-by: Michael Hwang <mhwang@research.att.com>
Change-Id: I67aa9178b1b1830e330ca1259e8f6b30202945df

5 files changed, 735 insertions, 0 deletions

diff --git a/mod/distributorapi/tests/components.json b/mod/distributorapi/tests/components.json
new file mode 100644
index 0000000..f9a71f2
--- /dev/null
+++ b/mod/distributorapi/tests/components.json
@@ -0,0 +1,260 @@
+[
+{
+  "spec": {
+    "self": {
+      "version": "1.5.0",
+      "name": "dcae-ves-collector",
+      "description": "Collector for receiving VES events through restful interface",
+      "component_type": "docker"
+    },
+    "streams": {
+      "subscribes": [],
+      "publishes": [
+        {
+          "format": "VES_specification",
+          "version": "7.30.1",
+          "type": "message router",
+          "config_key": "ves-pnfRegistration"
+        },
+        {
+          "format": "VES_specification",
+          "version": "7.30.1",
+          "type": "message router",
+          "config_key": "ves-pnfRegistration-secondary"
+        },
+        {
+          "format": "VES_specification",
+          "version": "7.30.1",
+          "type": "message router",
+          "config_key": "ves-notification"
+        },
+        {
+          "format": "VES_specification",
+          "version": "7.30.1",
+          "type": "message router",
+          "config_key": "ves-notification-secondary"
+        }
+      ]
+    },
+    "services": {
+      "calls": [],
+      "provides": []
+    },
+    "parameters": [
+      {
+        "name": "collector.service.port",
+        "value": 8080,
+        "description": "standard http port collector will open for listening;",
+        "sourced_at_deployment": false,
+        "policy_editable": false,
+        "designer_editable": false
+      },
+      {
+        "name": "collector.service.secure.port",
+        "value": 8443,
+        "description": "secure http port collector will open for listening ",
+        "sourced_at_deployment": false,
+        "policy_editable": false,
+        "designer_editable": true
+      },
+      {
+        "name": "collector.keystore.file.location",
+        "value": "/opt/app/dcae-certificate/keystore.jks",
+        "description": "fs location of keystore file in vm",
+        "sourced_at_deployment": false,
+        "policy_editable": false,
+        "designer_editable": false
+      },
+      {
+        "name": "collector.keystore.passwordfile",
+        "value": "/opt/app/dcae-certificate/.password",
+        "description": "location of keystore password file in vm",
+        "sourced_at_deployment": false,
+        "policy_editable": false,
+        "designer_editable": false
+      },
+      {
+        "name": "collector.truststore.file.location",
+        "value": "/opt/app/dcae-certificate/truststore.jks",
+        "description": "fs location of truststore file in vm",
+        "sourced_at_deployment": false,
+        "policy_editable": false,
+        "designer_editable": false
+      },
+      {
+        "name": "collector.truststore.passwordfile",
+        "value": "/opt/app/dcae-certificate/.trustpassword",
+        "description": "location of truststore password file in vm",
+        "sourced_at_deployment": false,
+        "policy_editable": false,
+        "designer_editable": false
+      },
+      {
+        "name": "collector.dmaap.streamid",
+        "value": "fault=ves-fault,ves-fault-secondary|syslog=ves-syslog,ves-syslog-secondary|heartbeat=ves-heartbeat,ves-heartbeat-secondary|measurementsForVfScaling=ves-measurement,ves-measurement-secondary|mobileFlow=ves-mobileflow,ves-mobileflow-secondary|other=ves-other,ves-other-secondary|stateChange=ves-statechange,ves-statechange-secondary|thresholdCrossingAlert=ves-thresholdCrossingAlert,ves-thresholdCrossingAlert-secondary|voiceQuality=ves-voicequality,ves-voicequality-secondary|sipSignaling=ves-sipsignaling,ves-sipsignaling-secondary|notification=ves-notification,ves-notification-secondary|pnfRegistration=ves-pnfRegistration,ves-pnfRegistration-secondary",
+        "description": "domain-to-streamid mapping used by VESCollector to distributes events based on domain. Both primary and secondary config_key are included for resilency (multiple streamid can be included commma separated). The streamids MUST match to topic config_keys. For single site without resiliency deployment - configkeys with -secondary suffix can be removed",
+        "sourced_at_deployment": true,
+        "policy_editable": false,
+        "designer_editable": false
+      },
+      {
+        "name": "auth.method",
+        "value": "noAuth",
+        "description": "Property to manage application mode, possible configurations: noAuth - default option - no security (http) , certOnly - auth by certificate (https), basicAuth - auth by basic auth username and password (https),certBasicAuth - auth by certificate and basic auth username / password (https),",
+        "sourced_at_deployment": false,
+        "policy_editable": false,
+        "designer_editable": false
+      },
+      {
+        "name": "header.authlist",
+        "value": "sample1,$2a$10$pgjaxDzSuc6XVFEeqvxQ5u90DKJnM/u7TJTcinAlFJVaavXMWf/Zi|userid1,$2a$10$61gNubgJJl9lh3nvQvY9X.x4e5ETWJJ7ao7ZhJEvmfJigov26Z6uq|userid2,$2a$10$G52y/3uhuhWAMy.bx9Se8uzWinmbJa.dlm1LW6bYPdPkkywLDPLiy",
+        "description": "List of id and base 64 encoded password.For each onboarding VNF - unique userid and password should be assigned and communicated to VNF owner. Password value should be base64 encoded in config here",
+        "policy_editable": false,
+        "sourced_at_deployment": true,
+        "designer_editable": true
+      },
+      {
+        "name": "collector.schema.checkflag",
+        "value": 1,
+        "description": "Schema check validation flag. When enabled, collector will validate input VES events against VES Schema defined on collector.schema.file ",
+        "sourced_at_deployment": false,
+        "policy_editable": false,
+        "designer_editable": false
+      },
+      {
+        "name": "collector.schema.file",
+        "value": "{\"v1\":\"./etc/CommonEventFormat_27.2.json\",\"v2\":\"./etc/CommonEventFormat_27.2.json\",\"v3\":\"./etc/CommonEventFormat_27.2.json\",\"v4\":\"./etc/CommonEventFormat_27.2.json\",\"v5\":\"./etc/CommonEventFormat_28.4.1.json\",\"v7\":\"./etc/CommonEventFormat_30.json\"}",
+        "description": "VES schema file name per version used for validation",
+        "designer_editable": true,
+        "sourced_at_deployment": false,
+        "policy_editable": false
+      },
+      {
+        "name": "event.transform.flag",
+        "value": 1,
+        "description": "flag to enable tranformation rules defined under eventTransform.json; this is applicable when event tranformation rules preset should be activated for transforming <VES5.4 events to 5.4",
+        "sourced_at_deployment": false,
+        "policy_editable": false,
+        "designer_editable": false
+      },
+      {
+        "name": "tomcat.maxthreads",
+        "value": "200",
+        "description": "Tomcat control for concurrent request",
+        "sourced_at_deployment": false,
+        "policy_editable": false,
+        "designer_editable": false
+      }
+    ],
+    "auxilary": {
+      "healthcheck": {
+        "type": "https",
+        "interval": "15s",
+        "timeout": "1s",
+        "endpoint": "/healthcheck"
+      },
+      "volumes": [
+        {
+          "container": {
+            "bind": "/opt/app/dcae-certificate"
+          },
+          "host": {
+            "path": "/opt/app/dcae-certificate"
+          }
+        },
+        {
+          "container": {
+            "bind": "/opt/app/VESCollector/logs"
+          },
+          "host": {
+            "path": "/opt/logs/DCAE/VESCollector/logs"
+          }
+        },
+        {
+          "container": {
+            "bind": "/opt/app/VESCollector/etc"
+          },
+          "host": {
+            "path": "/opt/logs/DCAE/VESCollector/etc"
+          }
+        }
+      ],
+      "ports": [
+        "8080:8080",
+        "8443:8443"
+      ]
+    },
+    "artifacts": [
+      {
+        "type": "docker image",
+        "uri": "nexus3.onap.org:10001/onap/org.onap.dcaegen2.collectors.ves.vescollector:latest"
+      }
+    ]
+  },
+  "id": "75c9a179-b36b-4985-9445-d44c8768d6eb",
+  "name": "dcae-ves-collector",
+  "version": "1.5.0",
+  "owner": "spike",
+  "whenAdded": "2019-10-04T15:16:01.717861",
+  "modified": "2019-10-04T15:16:37.234517",
+  "status": "published",
+  "description": "Collector for receiving VES events through restful interface",
+  "componentType": "docker",
+  "componentUrl": "http://t3-2/onboarding/components/75c9a179-b36b-4985-9445-d44c8768d6eb"
+},
+{
+  "spec": {
+    "parameters": [],
+    "artifacts": [
+      {
+        "type": "docker image",
+        "uri": "anomaly-detector:3"
+      }
+    ],
+    "self": {
+      "version": "1.0.0",
+      "name": "anomaly-detector",
+      "component_type": "docker",
+      "description": "Automatically generated from Acumos model"
+    },
+    "streams": {
+      "publishes": [
+        {
+          "type": "message_router",
+          "version": "1.0.0",
+          "config_key": "predict_publisher",
+          "format": "OutputFormat"
+        }
+      ],
+      "subscribes": [
+        {
+          "type": "message_router",
+          "version": "1.0.0",
+          "config_key": "predict_subscriber",
+          "format": "PredictIn"
+        }
+      ]
+    },
+    "auxilary": {
+      "healthcheck": {
+        "endpoint": "/healthcheck",
+        "type": "http"
+      }
+    },
+    "services": {
+      "provides": [],
+      "calls": []
+    }
+  },
+  "id": "3fadb641-2079-4ca9-bb07-0df5952967fc",
+  "name": "anomaly-detector",
+  "version": "1.0.0",
+  "owner": "spike",
+  "whenAdded": "2019-10-04T18:00:56.433976",
+  "modified": "2019-10-04T18:00:56.433971",
+  "status": "unpublished",
+  "description": "Automatically generated from Acumos model",
+  "componentType": "docker",
+  "componentUrl": "http://t3-2/onboarding/components/3fadb641-2079-4ca9-bb07-0df5952967fc"
+}
+]
diff --git a/mod/distributorapi/tests/flow.json b/mod/distributorapi/tests/flow.json
new file mode 100644
index 0000000..63cf0f2
--- /dev/null
+++ b/mod/distributorapi/tests/flow.json
@@ -0,0 +1,305 @@
+{
+  "bucket": {
+    "allowBundleRedeploy": false,
+    "createdTimestamp": 1570196205153,
+    "identifier": "f645b971-f096-485c-9699-93a193d9c0fa",
+    "link": {
+      "href": "buckets/f645b971-f096-485c-9699-93a193d9c0fa",
+      "params": {
+        "rel": "self"
+      }
+    },
+    "name": "Demos",
+    "permissions": {
+      "canDelete": true,
+      "canRead": true,
+      "canWrite": true
+    }
+  },
+  "flow": {
+    "bucketIdentifier": "f645b971-f096-485c-9699-93a193d9c0fa",
+    "bucketName": "Demos",
+    "createdTimestamp": 1570589769648,
+    "description": "",
+    "identifier": "ddec92e2-97c3-4b41-80d7-fd38bbf5bb69",
+    "link": {
+      "href": "buckets/f645b971-f096-485c-9699-93a193d9c0fa/flows/ddec92e2-97c3-4b41-80d7-fd38bbf5bb69",
+      "params": {
+        "rel": "self"
+      }
+    },
+    "modifiedTimestamp": 1570673258622,
+    "name": "work-in-progress",
+    "type": "Flow",
+    "versionCount": 4
+  },
+  "flowContents": {
+    "comments": "",
+    "componentType": "PROCESS_GROUP",
+    "connections": [
+      {
+        "backPressureDataSizeThreshold": "1 GB",
+        "backPressureObjectThreshold": 10000,
+        "bends": [],
+        "componentType": "CONNECTION",
+        "destination": {
+          "comments": "",
+          "groupId": "07fc9b5e-bbcd-3487-aded-eeced74fdfaa",
+          "id": "a8134467-b4b4-348f-8a1c-8d732fe4fcad",
+          "name": "AnomalyDetector",
+          "type": "PROCESSOR"
+        },
+        "flowFileExpiration": "0 sec",
+        "groupIdentifier": "07fc9b5e-bbcd-3487-aded-eeced74fdfaa",
+        "identifier": "04ab8849-858d-36d5-b7cf-40da26051759",
+        "labelIndex": 1,
+        "loadBalanceCompression": "DO_NOT_COMPRESS",
+        "loadBalanceStrategy": "DO_NOT_LOAD_BALANCE",
+        "name": "foo-conn",
+        "partitioningAttribute": "",
+        "prioritizers": [],
+        "selectedRelationships": [
+          "publishes:ves_specification:7.30.1:message router:ves-pnfRegistration-secondary",
+          "subscribes:predictin:1.0.0:message_router:predict_subscriber"
+        ],
+        "source": {
+          "comments": "",
+          "groupId": "07fc9b5e-bbcd-3487-aded-eeced74fdfaa",
+          "id": "3a0b5a44-f262-3836-93d5-dfe45fd90996",
+          "name": "DcaeVesCollector",
+          "type": "PROCESSOR"
+        },
+        "zIndex": 0
+      },
+      {
+        "backPressureDataSizeThreshold": "1 GB",
+        "backPressureObjectThreshold": 10000,
+        "bends": [],
+        "componentType": "CONNECTION",
+        "destination": {
+          "comments": "",
+          "groupId": "07fc9b5e-bbcd-3487-aded-eeced74fdfaa",
+          "id": "3a0b5a44-f262-3836-93d5-dfe45fd90996",
+          "name": "DcaeVesCollector",
+          "type": "PROCESSOR"
+        },
+        "flowFileExpiration": "0 sec",
+        "groupIdentifier": "07fc9b5e-bbcd-3487-aded-eeced74fdfaa",
+        "identifier": "c7c62954-ae06-388c-a6d9-a24243b51bea",
+        "labelIndex": 1,
+        "loadBalanceCompression": "DO_NOT_COMPRESS",
+        "loadBalanceStrategy": "DO_NOT_LOAD_BALANCE",
+        "name": "ves-data-conn",
+        "partitioningAttribute": "",
+        "prioritizers": [],
+        "selectedRelationships": [
+          "subscribes:ves_specification:7.30.1:message router:ves-notification"
+        ],
+        "source": {
+          "comments": "",
+          "groupId": "07fc9b5e-bbcd-3487-aded-eeced74fdfaa",
+          "id": "58629a8a-e813-3bb3-ac2d-7ce2f4abf604",
+          "name": "ves-notification",
+          "type": "INPUT_PORT"
+        },
+        "zIndex": 0
+      }
+    ],
+    "controllerServices": [],
+    "funnels": [],
+    "identifier": "07fc9b5e-bbcd-3487-aded-eeced74fdfaa",
+    "inputPorts": [
+      {
+        "allowRemoteAccess": false,
+        "comments": "",
+        "componentType": "INPUT_PORT",
+        "concurrentlySchedulableTaskCount": 1,
+        "groupIdentifier": "07fc9b5e-bbcd-3487-aded-eeced74fdfaa",
+        "identifier": "58629a8a-e813-3bb3-ac2d-7ce2f4abf604",
+        "name": "ves-notification",
+        "position": {
+          "x": 1168.5,
+          "y": 544.5
+        },
+        "type": "INPUT_PORT"
+      }
+    ],
+    "labels": [],
+    "name": "test",
+    "outputPorts": [],
+    "position": {
+      "x": 826,
+      "y": 301
+    },
+    "processGroups": [],
+    "processors": [
+      {
+        "autoTerminatedRelationships": [],
+        "bulletinLevel": "WARN",
+        "bundle": {
+          "artifact": "anomaly-detector",
+          "group": "org.onap.dcae",
+          "version": "1.0.0"
+        },
+        "comments": "",
+        "componentType": "PROCESSOR",
+        "concurrentlySchedulableTaskCount": 1,
+        "executionNode": "ALL",
+        "groupIdentifier": "07fc9b5e-bbcd-3487-aded-eeced74fdfaa",
+        "identifier": "a8134467-b4b4-348f-8a1c-8d732fe4fcad",
+        "name": "AnomalyDetector",
+        "penaltyDuration": "30 sec",
+        "position": {
+          "x": 2410,
+          "y": 824
+        },
+        "properties": {},
+        "propertyDescriptors": {},
+        "runDurationMillis": 0,
+        "schedulingPeriod": "0 sec",
+        "schedulingStrategy": "TIMER_DRIVEN",
+        "style": {},
+        "type": "org.onap.dcae.AnomalyDetector",
+        "yieldDuration": "1 sec"
+      },
+      {
+        "autoTerminatedRelationships": [],
+        "bulletinLevel": "WARN",
+        "bundle": {
+          "artifact": "dcae-ves-collector",
+          "group": "org.onap.dcae",
+          "version": "1.5.0"
+        },
+        "comments": "",
+        "componentType": "PROCESSOR",
+        "concurrentlySchedulableTaskCount": 1,
+        "executionNode": "ALL",
+        "groupIdentifier": "07fc9b5e-bbcd-3487-aded-eeced74fdfaa",
+        "identifier": "3a0b5a44-f262-3836-93d5-dfe45fd90996",
+        "name": "DcaeVesCollector",
+        "penaltyDuration": "30 sec",
+        "position": {
+          "x": 1500,
+          "y": 690
+        },
+        "properties": {
+          "collector.keystore.passwordfile": "/opt/app/dcae-certificate/.password",
+          "event.transform.flag": "1",
+          "collector.service.port": "8080",
+          "collector.schema.checkflag": "1",
+          "tomcat.maxthreads": "200",
+          "collector.truststore.passwordfile": "/opt/app/dcae-certificate/.trustpassword",
+          "header.authlist": "sample1,$2a$10$pgjaxDzSuc6XVFEeqvxQ5u90DKJnM/u7TJTcinAlFJVaavXMWf/Zi|userid1,$2a$10$61gNubgJJl9lh3nvQvY9X.x4e5ETWJJ7ao7ZhJEvmfJigov26Z6uq|userid2,$2a$10$G52y/3uhuhWAMy.bx9Se8uzWinmbJa.dlm1LW6bYPdPkkywLDPLiy",
+          "collector.service.secure.port": "8443",
+          "collector.keystore.file.location": "/opt/app/dcae-certificate/keystore.jks",
+          "collector.dmaap.streamid": "fault=ves-fault,ves-fault-secondary|syslog=ves-syslog,ves-syslog-secondary|heartbeat=ves-heartbeat,ves-heartbeat-secondary|measurementsForVfScaling=ves-measurement,ves-measurement-secondary|mobileFlow=ves-mobileflow,ves-mobileflow-secondary|other=ves-other,ves-other-secondary|stateChange=ves-statechange,ves-statechange-secondary|thresholdCrossingAlert=ves-thresholdCrossingAlert,ves-thresholdCrossingAlert-secondary|voiceQuality=ves-voicequality,ves-voicequality-secondary|sipSignaling=ves-sipsignaling,ves-sipsignaling-secondary|notification=ves-notification,ves-notification-secondary|pnfRegistration=ves-pnfRegistration,ves-pnfRegistration-secondary",
+          "collector.schema.file": "{\"v1\":\"./etc/CommonEventFormat_27.2.json\",\"v2\":\"./etc/CommonEventFormat_27.2.json\",\"v3\":\"./etc/CommonEventFormat_27.2.json\",\"v4\":\"./etc/CommonEventFormat_27.2.json\",\"v5\":\"./etc/CommonEventFormat_28.4.1.json\",\"v7\":\"./etc/CommonEventFormat_30.json\"}",
+          "auth.method": "noAuth",
+          "collector.truststore.file.location": "/opt/app/dcae-certificate/truststore.jks"
+        },
+        "propertyDescriptors": {
+          "collector.keystore.passwordfile": {
+            "displayName": "collector.keystore.passwordfile",
+            "identifiesControllerService": false,
+            "name": "collector.keystore.passwordfile",
+            "sensitive": false
+          },
+          "event.transform.flag": {
+            "displayName": "event.transform.flag",
+            "identifiesControllerService": false,
+            "name": "event.transform.flag",
+            "sensitive": false
+          },
+          "collector.service.port": {
+            "displayName": "collector.service.port",
+            "identifiesControllerService": false,
+            "name": "collector.service.port",
+            "sensitive": false
+          },
+          "collector.schema.checkflag": {
+            "displayName": "collector.schema.checkflag",
+            "identifiesControllerService": false,
+            "name": "collector.schema.checkflag",
+            "sensitive": false
+          },
+          "tomcat.maxthreads": {
+            "displayName": "tomcat.maxthreads",
+            "identifiesControllerService": false,
+            "name": "tomcat.maxthreads",
+            "sensitive": false
+          },
+          "collector.truststore.passwordfile": {
+            "displayName": "collector.truststore.passwordfile",
+            "identifiesControllerService": false,
+            "name": "collector.truststore.passwordfile",
+            "sensitive": false
+          },
+          "header.authlist": {
+            "displayName": "header.authlist",
+            "identifiesControllerService": false,
+            "name": "header.authlist",
+            "sensitive": false
+          },
+          "collector.service.secure.port": {
+            "displayName": "collector.service.secure.port",
+            "identifiesControllerService": false,
+            "name": "collector.service.secure.port",
+            "sensitive": false
+          },
+          "collector.keystore.file.location": {
+            "displayName": "collector.keystore.file.location",
+            "identifiesControllerService": false,
+            "name": "collector.keystore.file.location",
+            "sensitive": false
+          },
+          "collector.dmaap.streamid": {
+            "displayName": "collector.dmaap.streamid",
+            "identifiesControllerService": false,
+            "name": "collector.dmaap.streamid",
+            "sensitive": false
+          },
+          "collector.schema.file": {
+            "displayName": "collector.schema.file",
+            "identifiesControllerService": false,
+            "name": "collector.schema.file",
+            "sensitive": false
+          },
+          "auth.method": {
+            "displayName": "auth.method",
+            "identifiesControllerService": false,
+            "name": "auth.method",
+            "sensitive": false
+          },
+          "collector.truststore.file.location": {
+            "displayName": "collector.truststore.file.location",
+            "identifiesControllerService": false,
+            "name": "collector.truststore.file.location",
+            "sensitive": false
+          }
+        },
+        "runDurationMillis": 0,
+        "schedulingPeriod": "0 sec",
+        "schedulingStrategy": "TIMER_DRIVEN",
+        "style": {},
+        "type": "org.onap.dcae.DcaeVesCollector",
+        "yieldDuration": "1 sec"
+      }
+    ],
+    "remoteProcessGroups": [],
+    "variables": {}
+  },
+  "snapshotMetadata": {
+    "author": "anonymous",
+    "bucketIdentifier": "f645b971-f096-485c-9699-93a193d9c0fa",
+    "comments": "",
+    "flowIdentifier": "ddec92e2-97c3-4b41-80d7-fd38bbf5bb69",
+    "link": {
+      "href": "buckets/f645b971-f096-485c-9699-93a193d9c0fa/flows/ddec92e2-97c3-4b41-80d7-fd38bbf5bb69/versions/4",
+      "params": {
+        "rel": "content"
+      }
+    },
+    "timestamp": 1570673258587,
+    "version": 4
+  }
+}
diff --git a/mod/distributorapi/tests/test_registry_client.py b/mod/distributorapi/tests/test_registry_client.py
new file mode 100644
index 0000000..3767242
--- /dev/null
+++ b/mod/distributorapi/tests/test_registry_client.py
@@ -0,0 +1,78 @@
+# ============LICENSE_START=======================================================
+# Copyright (c) 2019 AT&T Intellectual Property. All rights reserved.
+# ================================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=========================================================
+import distributor
+from distributor import registry_client as rc
+
+
+def test_add_url_from_link():
+    test = {"link": {"href": "bar"}, "name": "jane", "age": 33,
+            "innerTest": {"link": {"href": "baz"}, "name": "bob"}
+            }
+    result = rc._add_url_from_link("http://foo", test)
+
+    assert result["selfUrl"] == "http://foo/bar"
+    assert result["innerTest"]["selfUrl"] == "http://foo/baz"
+
+
+def test_get_buckets(monkeypatch):
+    def fake_get_json(url):
+        if url == "http://registry/buckets":
+            return []
+        return None
+
+    monkeypatch.setattr(distributor.registry_client, "_get_json", fake_get_json)
+
+    assert [] == rc.get_buckets("http://registry")
+    assert [] == rc.get_buckets("http://registry/")
+
+
+def test_find_flow(monkeypatch):
+    def fake_get_buckets(url):
+        return [{"selfUrl": "{0}/buckets/123".format(url)}]
+
+    monkeypatch.setattr(distributor.registry_client, "get_buckets", fake_get_buckets)
+
+    def fake_get_flows(registry_url, url):
+        if url == "http://registry/buckets/123":
+            return [{"identifier": "abc"}]
+        return None
+
+    monkeypatch.setattr(distributor.registry_client, "get_flows", fake_get_flows)
+
+    assert rc.find_flow("http://registry", "abc")["identifier"] == "abc"
+
+
+def test_flow_versions(monkeypatch):
+    def fake_get_json_many(url):
+        if url == "http://registry/buckets/123/flows/abc/versions":
+            return [{"version": 1}, {"version": 3}, {"version": 2}]
+        print(url)
+        return []
+
+    monkeypatch.setattr(distributor.registry_client, "_get_json",
+            fake_get_json_many)
+
+    assert [3, 2, 1] == rc.get_flow_versions("http://registry/buckets/123/flows/abc/")
+
+
+def test_get_flow_diff_latest(monkeypatch):
+    def fake_get_flow_versions(url):
+        return ["1"]
+
+    monkeypatch.setattr(distributor.registry_client, "get_flow_versions",
+            fake_get_flow_versions)
+
+    assert None == rc.get_flow_diff_latest("http://registry", "http://registry/buckets/123/flows/abc/")
diff --git a/mod/distributorapi/tests/test_transform.py b/mod/distributorapi/tests/test_transform.py
new file mode 100644
index 0000000..f275e63
--- /dev/null
+++ b/mod/distributorapi/tests/test_transform.py
@@ -0,0 +1,68 @@
+# ============LICENSE_START=======================================================
+# Copyright (c) 2019 AT&T Intellectual Property. All rights reserved.
+# ================================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=========================================================
+import os, json
+from distributor import transform as tr
+
+TEST_DIR = os.path.dirname(__file__)
+
+def _load_data(filename):
+    path = os.path.join(TEST_DIR, filename)
+    with open(path) as f:
+        return json.load(f)
+
+def _setup():
+    flow = _load_data("flow.json")
+    components = _load_data("components.json")
+    components = dict([((c["name"], c["version"]), c) for c in components])
+    return (flow, components)
+
+
+def test_get_component():
+    flow, components = _setup()
+    c = tr.get_component(flow, components, "a8134467-b4b4-348f-8a1c-8d732fe4fcad")
+    assert "3fadb641-2079-4ca9-bb07-0df5952967fc" == c["id"]
+
+
+def test_make_fbp_from_flow():
+    flow, components = _setup()
+
+    fbp = tr.make_fbp_from_flow(flow, components)
+    assert len(fbp) == 4
+
+    def check_node(n):
+        n["payload"]["component_id"]
+
+    expected = ["75c9a179-b36b-4985-9445-d44c8768d6eb", "3fadb641-2079-4ca9-bb07-0df5952967fc"]
+    actual = [n["payload"]["component_id"] for n in fbp if n["command"] == "addnode"]
+    assert list(sorted(expected)) == list(sorted(actual))
+
+    # Test processor to processor scenario
+    expected = {'metadata': {'data_type': 'json',
+                           'dmaap_type': 'MR',
+                           'name': 'foo-conn'},
+              'src': {'node': '75c9a179-b36b-4985-9445-d44c8768d6eb',
+                      'port': 'ves-pnfRegistration-secondary'},
+              'tgt': {'node': '3fadb641-2079-4ca9-bb07-0df5952967fc',
+                      'port': 'predict_subscriber'}}
+    actual = [e["payload"] for e in fbp if e["command"] == "addedge"]
+    assert actual[0] == expected or actual[1] == expected
+
+    # Test input port to processor scenario
+    expected = {'metadata': {'data_type': 'json', 'dmaap_type': 'MR',
+        'name': 'ves-data-conn'}, 'src': {},
+        'tgt': {'node': '75c9a179-b36b-4985-9445-d44c8768d6eb',
+            'port': 'ves-notification'}}
+    assert actual[0] == expected or actual[1] == expected
diff --git a/mod/distributorapi/tests/test_utils.py b/mod/distributorapi/tests/test_utils.py
new file mode 100644
index 0000000..481e325
--- /dev/null
+++ b/mod/distributorapi/tests/test_utils.py
@@ -0,0 +1,24 @@
+# ============LICENSE_START=======================================================
+# Copyright (c) 2019 AT&T Intellectual Property. All rights reserved.
+# ================================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=========================================================
+from distributor import utils
+
+
+def test_urljoin():
+    assert "http://foo/bar/baz" == utils.urljoin("http://foo", "bar", "baz")
+    assert "http://foo/bar/baz" == utils.urljoin("http://foo/", "bar", "baz")
+    assert "http://foo/bar/baz?name=some-name&version=1.5.0" \
+        == utils.urljoin("http://foo", "bar", "baz", **{"name": "some-name",
+            "version": "1.5.0"})