diff --git a/ci-scripts/cls_corenetwork.py b/ci-scripts/cls_corenetwork.py
new file mode 100644
index 0000000000..7bb508d86b
--- /dev/null
+++ b/ci-scripts/cls_corenetwork.py
@@ -0,0 +1,169 @@
+# * Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
+# * contributor license agreements.  See the NOTICE file distributed with
+# * this work for additional information regarding copyright ownership.
+# * The OpenAirInterface Software Alliance licenses this file to You under
+# * the OAI Public License, Version 1.1  (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.openairinterface.org/?page_id=698
+# *
+# * 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.
+# *-------------------------------------------------------------------------------
+# * For more information about the OpenAirInterface (OAI) Software Alliance:
+# *      contact@openairinterface.org
+# */
+#---------------------------------------------------------------------
+#
+#   Required Python Version
+#     Python 3.x
+#
+#---------------------------------------------------------------------
+
+import logging
+import yaml
+import re
+
+import cls_cmd
+
+def listify(s):
+	if s is None:
+		return None
+	if isinstance(s, list):
+		return [str(x) for x in s]
+	return [str(s)]
+
+class CoreNetwork:
+
+	def __init__(self, cn_name, node=None, d=None, filename="ci_infra.yaml"):
+		with open(filename, "r") as f:
+			all_cns = yaml.load(f, Loader=yaml.FullLoader) # TODO why full?
+		c = all_cns.get(cn_name)
+		if c is None:
+			raise Exception(f'no such core network name "{cn_name}" in "{filename}"')
+		self._cn_name = cn_name
+		self._host = c.get('Host').strip()
+		if self._host == "%%current_host%%":
+			if node is None:
+				raise Exception(f"core network {self} requires node, but none provided (cannot replace %%current_host%%)")
+			self._host = node
+		if d is not None:
+			raise Exception("directory handling not implemented")
+		self._deploy = listify(c.get('Deploy'))
+		self._undeploy = listify(c.get('Undeploy'))
+		self._logCollect = listify(c.get('LogCollect'))
+		# check if at least one command contains "%%log_dir%%"
+		if self._logCollect and not any(filter(lambda s: "%%log_dir%%" in s, self._logCollect)):
+			raise ValueError(f"(At least one) LogCollect expression for {cn_name} must contain \"%%log_dir%%\"")
+		self._getNetwork = listify(c.get('NetworkScript'))
+		self._cmd_prefix = c.get('CmdPrefix')
+		iperf = c.get('RunIperf3Server')
+		self._run_iperf = bool(iperf if iperf is not None else True)
+		logging.info(f'initialized core {self} from {filename}')
+
+	def __str__(self):
+		return f"{self._cn_name}@{self._host} [IP: {self.getIP()}]"
+
+	def __repr__(self):
+		return self.__str__()
+
+	def _exec_script(host, line, silent):
+		# take off "!", split in words
+		words = line[1:].strip().split(" ")
+		script_name = words[0]
+		options = " ".join(words[1:])
+		ret = cls_cmd.runScript(host, script_name, 300, parameters=options, silent=silent)
+		return ret
+
+	def _command(self, cmd_list, must_succeed=False, silent=False):
+		succeeded = True
+		output = ""
+		with cls_cmd.getConnection(self._host) as c:
+			for cmd in cmd_list:
+				cmd = cmd.strip()
+				if cmd.startswith("!"):
+					ret = CoreNetwork._exec_script(self._host, cmd, silent=silent)
+				else:
+					ret = c.run(cmd, silent=silent)
+				output += ret.stdout
+				if ret.returncode != 0:
+					logging.warn(f"cmd \"{cmd}\" returned code {ret.returncode}, stdout {ret.stdout}")
+				if ret.returncode != 0 and must_succeed:
+					succeeded = False
+					break
+		return succeeded, output
+
+	def deploy(self):
+		# we first undeploy to make sure the core has been stopped
+		logging.info(f'undeploy core network {self} before deployment')
+		self._command(self._undeploy)
+		logging.info(f'deploy core network {self}')
+		success, output = self._command(self._deploy, must_succeed=True)
+		if not success:
+			logging.error(f'failure during deployment of core network {self}:\n{output}')
+			return False, output
+		logging.info(f'retrieve IP address')
+		ip = self.getIP()
+		if ip is None:
+			logging.error(f'could not retrieve pingable address of core network {self}')
+			return False, output
+		logging.info(f'deployed core network {self}, pingable IP address {ip}')
+		return True, output
+
+	def _collect_logs(self, log_dir):
+		logging.info(f'collecting logs into (local) {log_dir}')
+		remote_dir = "/tmp/cn-undeploy-logs"
+		with cls_cmd.getConnection(self._host) as c:
+			# create a directory for log collection
+			c.run(f'rm -rf {remote_dir}')
+			ret = c.run(f'mkdir {remote_dir}')
+			if ret.returncode != 0:
+				logging.error("cannot create directory for log collection")
+				return []
+			# inform core network implementation where to store logs
+			# (remote_dir), and trigger log collection
+			log_cmd = [s.replace('%%log_dir%%', remote_dir) for s in self._logCollect]
+			self._command(log_cmd)
+			# enumerate collected files
+			ret = c.run(f'ls {remote_dir}/*')
+			if ret.returncode != 0:
+				logging.error("cannot enumerate log files")
+				return []
+			log_files = []
+			# copy them to the executor one by one, and store in log_dir
+			for f in ret.stdout.split("\n"):
+				l = f.replace(remote_dir, log_dir)
+				c.copyin(f, l)
+				log_files.append(l)
+			c.run(f'rm -rf {remote_dir}')
+			return log_files
+
+	def undeploy(self, log_dir=None):
+		log_files = []
+		if log_dir is not None:
+			log_files = self._collect_logs(log_dir)
+		else:
+			logging.warning("no directory for log collection specified, cannot retrieve core network logs")
+		logging.info(f'undeploy core network {self}')
+		_, output = self._command(self._undeploy)
+		return log_files, output
+
+	def getIP(self):
+		success, output = self._command(self._getNetwork)
+		result = re.search(r'inet (?P<ip>[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)', output)
+		if success and result and result.group('ip'):
+			return result.group('ip')
+		return None
+
+	def getCmdPrefix(self):
+		return self._cmd_prefix or ""
+
+	def getHost(self):
+		return self._host
+
+	def runIperf3Server(self):
+		return self._run_iperf
diff --git a/ci-scripts/tests/README.md b/ci-scripts/tests/README.md
index 364cf031d3..f583c51c08 100644
--- a/ci-scripts/tests/README.md
+++ b/ci-scripts/tests/README.md
@@ -12,6 +12,7 @@ To run individual unit tests, start them like so:
 
     python tests/build.py -v
     python tests/cmd.py -v
+    python tests/corenetwork.py -v
     python tests/deployment.py -v
     python tests/iperf-analysis.py -v
     python tests/ping-iperf.py -v
diff --git a/ci-scripts/tests/config/test_core_infra.yaml b/ci-scripts/tests/config/test_core_infra.yaml
new file mode 100644
index 0000000000..dfa0b4817d
--- /dev/null
+++ b/ci-scripts/tests/config/test_core_infra.yaml
@@ -0,0 +1,46 @@
+test:
+  Host: localhost
+  NetworkScript: echo "inet 127.0.0.1 mtu 1500"
+  Deploy: echo deploy
+  Undeploy: echo undeploy
+  LogCollect: echo logs > %%log_dir%%/logs
+
+test_list:
+  Host: localhost
+  NetworkScript: echo "inet 127.0.0.1 mtu 1500"
+  Deploy:
+    - echo deploy
+    - echo multistep deploy!!
+  Undeploy:
+    - "echo undeploy"
+    - "echo terminated"
+  LogCollect:
+    - echo logs > %%log_dir%%/logs
+  RunIperf3Server: false
+
+test_fail:
+  Host: localhost
+  NetworkScript: echo "inet 127.0.0.1 mtu 1500"
+  Deploy:
+    - echo deploy
+    - "false"
+    - echo multistep deploy!!
+  Undeploy:
+    - "echo undeploy"
+    - "echo terminated"
+  LogCollect:
+    - echo logs > %%log_dir%%/logs
+
+test_script:
+  Host: localhost
+  NetworkScript: echo "inet 127.0.0.1 mtu 1500"
+  Deploy: ! tests/scripts/deploy-core.sh
+  Undeploy: "echo undeploy"
+  LogCollect:  echo logs > %%log_dir%%/logs
+
+test_script_fail:
+  Host: localhost
+  NetworkScript: echo "inet 127.0.0.1 mtu 1500"
+  Deploy: ! tests/scripts/deploy-core.sh WILLFAIL
+  Undeploy: "echo undeploy"
+  LogCollect:  echo logs > %%log_dir%%/logs
diff --git a/ci-scripts/tests/corenetwork.py b/ci-scripts/tests/corenetwork.py
new file mode 100644
index 0000000000..77e5c7629e
--- /dev/null
+++ b/ci-scripts/tests/corenetwork.py
@@ -0,0 +1,66 @@
+import sys
+import logging
+logging.basicConfig(
+	level=logging.DEBUG,
+	stream=sys.stdout,
+	format="[%(asctime)s] %(levelname)8s: %(message)s"
+)
+import os
+
+import unittest
+
+sys.path.append('./') # to find OAI imports below
+import cls_corenetwork
+import cls_cmd
+
+class TestCoreNetwork(unittest.TestCase):
+    def test_simple_core(self):
+        c = cls_corenetwork.CoreNetwork("test", filename="tests/config/test_core_infra.yaml")
+        success, output = c.deploy()
+        self.assertTrue(success)
+        self.assertEqual(output, "deploy")
+        # should not have it set in config
+        self.assertTrue(c.runIperf3Server())
+        self.assertEqual(c.getIP(), "127.0.0.1")
+        log_files, output = c.undeploy(log_dir="/tmp")
+        self.assertEqual(output, "undeploy")
+        with cls_cmd.LocalCmd() as cmd:
+            # there must be one log file for this (test) core
+            self.assertEqual(len(log_files), 1)
+            # test core writes to %%log_dir%%/logs, which expands to /tmp/logs
+            l = log_files[0]
+            self.assertEqual(l, "/tmp/logs")
+            ret = cmd.run(f"cat {l}")
+            self.assertEqual(ret.returncode, 0) # command must succeed
+            self.assertEqual(ret.stdout, "logs") # output should be "logs"
+
+    def test_core_list(self):
+        c = cls_corenetwork.CoreNetwork("test_list", filename="tests/config/test_core_infra.yaml")
+        success, _ = c.deploy()
+        self.assertTrue(success)
+        self.assertFalse(c.runIperf3Server())
+        c.undeploy(log_dir="/tmp")
+
+    def test_core_fail(self):
+        c = cls_corenetwork.CoreNetwork("test_fail", filename="tests/config/test_core_infra.yaml")
+        success, _ = c.deploy()
+        self.assertFalse(success)
+        # undeployment should still work
+        c.undeploy(log_dir="/tmp")
+
+    def test_core_script(self):
+        c = cls_corenetwork.CoreNetwork("test_script", filename="tests/config/test_core_infra.yaml")
+        success, output = c.deploy()
+        self.assertTrue(success)
+        self.assertEqual(output, "deployment from script")
+        c.undeploy()
+
+    def test_core_script_fail(self):
+        c = cls_corenetwork.CoreNetwork("test_script_fail", filename="tests/config/test_core_infra.yaml")
+        success, output = c.deploy()
+        self.assertFalse(success)
+        self.assertEqual(output, "deployment from script\nfailing")
+        c.undeploy()
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/ci-scripts/tests/scripts/deploy-core.sh b/ci-scripts/tests/scripts/deploy-core.sh
new file mode 100755
index 0000000000..215ee15f70
--- /dev/null
+++ b/ci-scripts/tests/scripts/deploy-core.sh
@@ -0,0 +1,12 @@
+#!/bin/bash
+
+function die() {
+	echo $@
+	exit 1
+}
+
+echo "deployment from script"
+
+[ $# -gt 0 ] && die "failing"
+
+exit 0