Repository: ovn-org/libovsdb
Pull request #175 (2021-06-22 10:48:06)

test: Use Dockertest to simplify integration tests
This simplifies the integration test flow in two ways.

1. Use dockertest - which provides an easy way to create test containers
from our test code
2. Use testify/suite to organize the test cases

As a result:
- We can remove the docker-compose file and Dockerfile
- go test ./.. just works

Fixes: #136 

Signed-off-by: Dave Tucker <dave@dtucker.co.uk>

coveralls: ## Pull Request Test Coverage Report for [Build 960345331](https://coveralls.io/builds/40775122)

* **0** of **0**   changed or added relevant lines in **0** files are covered.
* **156** unchanged lines in **1** file lost coverage.
* Overall coverage decreased (**-3.8%**) to **70.084%**

---


|  Files with Coverage Reduction | New Missed Lines | % |
| :-----|--------------|--: |
| [client/client.go](https://coveralls.io/builds/40775122/source?filename=client%2Fclient.go#L73) | 156 | 16.78% |
<!-- | **Total:** | **156** |  | -->

|  Totals | [![Coverage Status](https://coveralls.io/builds/40775122/badge)](https://coveralls.io/builds/40775122) |
| :-- | --: |
| Change from base [Build 950624558](https://coveralls.io/builds/40709421): |  -3.8% |
| Covered Lines: | 2844 |
| Relevant Lines: | 4058 |

---
##### 💛  - [Coveralls](https://coveralls.io)

dave-tucker: This is also pre-requisite to #172 as I want to have the ability to stop/start the server container (or send a SIGHUP to ovsdb-server) to see what happens to monitors when the server is restarted and to (maybe?) simulate a flaky connection...
dave-tucker: To reduce risk of flakes, I've changed ths from a fresh instance of OVS per test to a fresh instance of OVS per suite.
dave-tucker: @amorenoz comments addressed, and I also ensured that the integration tests provide code coverage too :+1: 
amorenoz: LGTM
amorenoz: removed the LGTM because my local build still failing
dave-tucker: @amorenoz ok removed `bench` from the `all` target and hopefully this should be good to go now :tada: 

Patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 2aa2a00..5218dcb 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -24,8 +24,8 @@ jobs:
     - name: Check out code into the Go module directory
       uses: actions/checkout@v2
 
-    - name: Build
-      run: make build-local
+    - name: Generate Code
+      run: make prebuild
 
     # This cannot be run in parallel because we require running go-generate first
     # We need to skip the go installation and caching to avoid "File exists" errors in tbe logs
@@ -37,20 +37,25 @@ jobs:
         skip-pkg-cache: true
         skip-build-cache: true
 
-    - name: Check Formatting
-      run: make fmt
+    - name: Build
+      run: make build
+
+    - name: Test
+      run: make test
 
     - name: Test
-      run: make test-local
+      run: make integration-test
 
-    - name: Upload Unit Test Coverage
-      uses: actions/upload-artifact@v2
+    - name: Generate coverage
+      run: make coverage
+
+    - name: Upload test coverage
+      uses: shogo82148/actions-goveralls@v1
       with:
-        path: unit.cov
-        name: unit.cov
+        path-to-profile: profile.cov
 
     - name: Benchmark
-      run: make bench-local
+      run: make bench
 
     - name: Restore Latest Main Benchmark
       id: old-benchmark
@@ -75,19 +80,12 @@ jobs:
     strategy:
       matrix:
         ovs_version:
-          - latest
           - 2.15.0
           - 2.14.0
           - 2.13.0
 
     runs-on: ubuntu-latest
-    services:
-      ovs:
-        image: libovsdb/ovs:${{ matrix.ovs_version }}
-        options: --tty
-        ports:
-          - 6640:6640
-
+    
     steps:
     - name: Set up Go 1.16
       uses: actions/setup-go@v1
@@ -99,36 +97,6 @@ jobs:
       uses: actions/checkout@v2
 
     - name: Integration Test
-      run: make integration-test-local
+      run: make integration-test
       env:
-        OVS_DB: tcp:127.0.0.1:6640
-
-    - name: Upload Integration Test Coverage
-      uses: actions/upload-artifact@v2
-      with:
-        path: integration.cov
-        name: integration.cov
-
-  coverage:
-    name: Upload Coverage
-    needs: [build, test]
-    runs-on: ubuntu-latest
-    steps:
-    - name: Set up Go 1.16
-      uses: actions/setup-go@v1
-      with:
-        go-version: 1.16
-      id: go
-    - name: Check out code into the Go module directory
-      uses: actions/checkout@v2
-
-    - name: Restore Coverage Files
-      uses: actions/download-artifact@v2
-
-    - name: Combine Reports
-      run: cat unit.cov/unit.cov <(tail -n +2 integration.cov/integration.cov) > profile.cov
-
-    - name: Send coverage
-      uses: shogo82148/actions-goveralls@v1
-      with:
-        path-to-profile: profile.cov
+        OVS_IMAGE_TAG: ${{ matrix.ovs_version }}
\ No newline at end of file
diff --git a/Dockerfile.dev b/Dockerfile.dev
deleted file mode 100644
index f045e63..0000000
--- a/Dockerfile.dev
+++ /dev/null
@@ -1,2 +0,0 @@
-FROM golang:1.16
-RUN curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin v1.39.0
diff --git a/Makefile b/Makefile
index 7945ea6..3537064 100644
--- a/Makefile
+++ b/Makefile
@@ -1,46 +1,47 @@
-.PHONY: all local test test-local integration-test-local install-deps lint fmt prebuild-local build-local
+.PHONY: all
+all: lint build test integration-test coverage
 
-all: test
-
-local: install-deps build-local fmt lint test-local bench-local
-
-prebuild-local:
+.PHONY: prebuild
+prebuild: 
 	@echo "+ $@"
 	@mkdir -p bin
 	@go build -v -o ./bin ./cmd/modelgen
 	@[ -f example/play_with_ovs/ovs.ovsschema ] || curl -o example/play_with_ovs/ovs.ovsschema https://raw.githubusercontent.com/openvswitch/ovs/v2.15.0/vswitchd/vswitch.ovsschema
 	@go generate -v ./...
 
-build-local: prebuild-local
+.PHONY: build
+build: prebuild 
 	@echo "+ $@"
 	@go build -v ./...
 
-test-local:
+.PHONY: test
+test:
 	@echo "+ $@"
-	@go test -race -coverprofile=unit.cov -short -v ./...
+	@go test -race -coverprofile=unit.cov -test.short -timeout 30s -v ./...
 
-bench-local:
+.PHONY: integration-test
+integration-test:
 	@echo "+ $@"
-	@go test -run=XXX -count=3 -bench=. ./... | tee bench.out
-	@benchstat bench.out
+	@go test -race -coverprofile=integration.cov -coverpkg=github.com/ovn-org/libovsdb/... -timeout 60s -v ./test/ovs
 
-integration-test-local: prebuild-local
-	@echo "+ $@"
-	@go test -race -v -coverprofile=integration.cov -run Integration ./...
+.PHONY: coverage
+coverage: test integration-test
+	@sed -i '1d' integration.cov
+	@cat unit.cov integration.cov > profile.cov
 
-test:
-	@docker-compose pull
-	@docker-compose run --rm test
+.PHONY: bench
+bench: install-deps
+	@echo "+ $@"
+	@go test -run=XXX -count=3 -bench=. ./... | tee bench.out
+	@benchstat bench.out
 
+.PHONY: install-deps
 install-deps:
 	@echo "+ $@"
 	@golangci-lint --version
 	@go install golang.org/x/perf/cmd/benchstat@latest
 
-lint:
+.PHONY: lint
+lint: install-deps prebuild
 	@echo "+ $@"
 	@golangci-lint run
-
-fmt:
-	@echo "+ $@"
-	@test -z "$$(gofmt -s -l . | tee /dev/stderr)"
diff --git a/client/client.go b/client/client.go
index 7aa406e..660f411 100644
--- a/client/client.go
+++ b/client/client.go
@@ -60,10 +60,10 @@ type ovsdbClient struct {
 	schema     *ovsdb.DatabaseSchema
 	cache      *cache.TableCache
 	stopCh     chan struct{}
-	connected  bool
 	disconnect chan struct{}
 	api        API
-	mutex      sync.Mutex
+	// mutex protects the rpcClient from concurrent access
+	mutex sync.Mutex
 }
 
 // NewOVSDBClient creates a new OVSDB Client with the provided
@@ -93,11 +93,11 @@ func newOVSDBClient(databaseModel *model.DBModel, opts ...Option) (*ovsdbClient,
 // The connection can be configured using one or more Option(s), like WithTLSConfig
 // If no WithEndpoint option is supplied, the default of unix:/var/run/openvswitch/ovsdb.sock is used
 func (o *ovsdbClient) Connect(ctx context.Context) error {
-	if o.connected {
-		return nil
-	}
 	o.mutex.Lock()
 	defer o.mutex.Unlock()
+	if o.rpcClient != nil {
+		return nil
+	}
 	var c net.Conn
 	var dialer net.Dialer
 	var err error
@@ -132,24 +132,6 @@ func (o *ovsdbClient) Connect(ctx context.Context) error {
 	if err := o.createRPC2Client(c); err != nil {
 		return err
 	}
-	o.connected = true
-	return nil
-}
-
-// createRPC2Client creates an rpcClient using the provided connection
-// It is also responsible for setting up go routines for handling disconnect notification
-// and cache population
-func (o *ovsdbClient) createRPC2Client(conn net.Conn) error {
-	o.stopCh = make(chan struct{})
-	o.rpcClient = rpc2.NewClientWithCodec(jsonrpc.NewJSONCodec(conn))
-	o.rpcClient.SetBlocking(true)
-	o.rpcClient.Handle("echo", func(_ *rpc2.Client, args []interface{}, reply *[]interface{}) error {
-		return o.echo(args, reply)
-	})
-	o.rpcClient.Handle("update", func(_ *rpc2.Client, args []json.RawMessage, reply *[]interface{}) error {
-		return o.update(args, reply)
-	})
-	go o.rpcClient.Run()
 
 	dbs, err := o.listDbs()
 	if err != nil {
@@ -193,10 +175,25 @@ func (o *ovsdbClient) createRPC2Client(conn net.Conn) error {
 		o.rpcClient.Close()
 		return err
 	}
-
 	go o.cache.Run(o.stopCh)
-	go o.handleDisconnectNotification()
+	return nil
+}
 
+// createRPC2Client creates an rpcClient using the provided connection
+// It is also responsible for setting up go routines for client-side event handling
+// and handling disconnects. Should only be called when the mutex is held
+func (o *ovsdbClient) createRPC2Client(conn net.Conn) error {
+	o.stopCh = make(chan struct{})
+	o.rpcClient = rpc2.NewClientWithCodec(jsonrpc.NewJSONCodec(conn))
+	o.rpcClient.SetBlocking(true)
+	o.rpcClient.Handle("echo", func(_ *rpc2.Client, args []interface{}, reply *[]interface{}) error {
+		return o.echo(args, reply)
+	})
+	o.rpcClient.Handle("update", func(_ *rpc2.Client, args []json.RawMessage, reply *[]interface{}) error {
+		return o.update(args, reply)
+	})
+	go o.rpcClient.Run()
+	go o.handleDisconnectNotification()
 	return nil
 }
 
@@ -216,7 +213,9 @@ func (o *ovsdbClient) Cache() *cache.TableCache {
 // SetOption sets a new value for an option.
 // It may only be called when the client is not connected
 func (o *ovsdbClient) SetOption(opt Option) error {
-	if o.connected {
+	o.mutex.Lock()
+	defer o.mutex.Unlock()
+	if o.rpcClient != nil {
 		return fmt.Errorf("cannot set option when client is connected")
 	}
 	return opt(o.options)
@@ -224,7 +223,9 @@ func (o *ovsdbClient) SetOption(opt Option) error {
 
 // Connected returns whether or not the client is currently connected to the server
 func (o *ovsdbClient) Connected() bool {
-	return o.connected
+	o.mutex.Lock()
+	defer o.mutex.Unlock()
+	return o.rpcClient != nil
 }
 
 // DisconnectNotify returns a channel which will notify the caller when the
@@ -262,6 +263,7 @@ func (o *ovsdbClient) update(args []json.RawMessage, reply *[]interface{}) error
 
 // getSchema returns the schema in use for the provided database name
 // RFC 7047 : get_schema
+// Should only be called when mutex is held
 func (o *ovsdbClient) getSchema(dbName string) (*ovsdb.DatabaseSchema, error) {
 	args := ovsdb.NewGetSchemaArgs(dbName)
 	var reply ovsdb.DatabaseSchema
@@ -274,6 +276,7 @@ func (o *ovsdbClient) getSchema(dbName string) (*ovsdb.DatabaseSchema, error) {
 
 // listDbs returns the list of databases on the server
 // RFC 7047 : list_dbs
+// Should only be called when mutex is held
 func (o *ovsdbClient) listDbs() ([]string, error) {
 	var dbs []string
 	err := o.rpcClient.Call("list_dbs", nil, &dbs)
@@ -286,9 +289,6 @@ func (o *ovsdbClient) listDbs() ([]string, error) {
 // Transact performs the provided Operations on the database
 // RFC 7047 : transact
 func (o *ovsdbClient) Transact(operation ...ovsdb.Operation) ([]ovsdb.OperationResult, error) {
-	if !o.connected {
-		return nil, ErrNotConnected
-	}
 	var reply []ovsdb.OperationResult
 
 	if ok := o.schema.ValidateOperations(operation...); !ok {
@@ -296,7 +296,13 @@ func (o *ovsdbClient) Transact(operation ...ovsdb.Operation) ([]ovsdb.OperationR
 	}
 
 	args := ovsdb.NewTransactArgs(o.schema.Name, operation...)
+	o.mutex.Lock()
+	if o.rpcClient == nil {
+		o.mutex.Unlock()
+		return nil, ErrNotConnected
+	}
 	err := o.rpcClient.Call("transact", args, &reply)
+	o.mutex.Unlock()
 	if err != nil {
 		return nil, err
 	}
@@ -305,9 +311,6 @@ func (o *ovsdbClient) Transact(operation ...ovsdb.Operation) ([]ovsdb.OperationR
 
 // MonitorAll is a convenience method to monitor every table/column
 func (o *ovsdbClient) MonitorAll(jsonContext interface{}) error {
-	if !o.connected {
-		return ErrNotConnected
-	}
 	var options []TableMonitor
 	for name := range o.dbModel.Types() {
 		options = append(options, TableMonitor{Table: name})
@@ -318,13 +321,13 @@ func (o *ovsdbClient) MonitorAll(jsonContext interface{}) error {
 // MonitorCancel will request cancel a previously issued monitor request
 // RFC 7047 : monitor_cancel
 func (o *ovsdbClient) MonitorCancel(jsonContext interface{}) error {
-	if !o.connected {
-		return ErrNotConnected
-	}
 	var reply ovsdb.OperationResult
-
 	args := ovsdb.NewMonitorCancelArgs(jsonContext)
-
+	o.mutex.Lock()
+	defer o.mutex.Unlock()
+	if o.rpcClient == nil {
+		return ErrNotConnected
+	}
 	err := o.rpcClient.Call("monitor_cancel", args, &reply)
 	if err != nil {
 		return err
@@ -364,9 +367,6 @@ func (o *ovsdbClient) NewTableMonitor(m model.Model, fields ...interface{}) Tabl
 // by the Update Notifications
 // RFC 7047 : monitor
 func (o *ovsdbClient) Monitor(jsonContext interface{}, options ...TableMonitor) error {
-	if !o.connected {
-		return ErrNotConnected
-	}
 	if len(options) == 0 {
 		return fmt.Errorf("no monitor options provided")
 	}
@@ -389,6 +389,11 @@ func (o *ovsdbClient) Monitor(jsonContext interface{}, options ...TableMonitor)
 		requests[o.Table] = *request
 	}
 	args := ovsdb.NewMonitorArgs(o.schema.Name, jsonContext, requests)
+	o.mutex.Lock()
+	defer o.mutex.Unlock()
+	if o.rpcClient == nil {
+		return ErrNotConnected
+	}
 	err := o.rpcClient.Call("monitor", args, &reply)
 	if err != nil {
 		return err
@@ -399,11 +404,13 @@ func (o *ovsdbClient) Monitor(jsonContext interface{}, options ...TableMonitor)
 
 // Echo tests the liveness of the OVSDB connetion
 func (o *ovsdbClient) Echo() error {
-	if !o.connected {
-		return ErrNotConnected
-	}
 	args := ovsdb.NewEchoArgs()
 	var reply []interface{}
+	o.mutex.Lock()
+	defer o.mutex.Unlock()
+	if o.rpcClient == nil {
+		return ErrNotConnected
+	}
 	err := o.rpcClient.Call("echo", args, &reply)
 	if err != nil {
 		return err
@@ -415,15 +422,11 @@ func (o *ovsdbClient) Echo() error {
 }
 
 func (o *ovsdbClient) handleDisconnectNotification() {
-	// this will block until Connect() has released the lock via defer
+	<-o.rpcClient.DisconnectNotify()
 	o.mutex.Lock()
-	// we continue to hold the lock until the client has disconnected
-	// this prevents another call to Connect() changing the rpcClient
-	// while we're still listening for disconnects
 	defer o.mutex.Unlock()
-	<-o.rpcClient.DisconnectNotify()
-	close(o.stopCh)
 	o.rpcClient = nil
+	close(o.stopCh)
 	o.cache = nil
 	select {
 	case o.disconnect <- struct{}{}:
@@ -435,10 +438,11 @@ func (o *ovsdbClient) handleDisconnectNotification() {
 
 // Disconnect will close the connection to the OVSDB server
 func (o *ovsdbClient) Disconnect() {
-	if !o.connected {
+	o.mutex.Lock()
+	defer o.mutex.Unlock()
+	if o.rpcClient == nil {
 		return
 	}
-	o.connected = false
 	o.rpcClient.Close()
 }
 
diff --git a/docker-compose.yml b/docker-compose.yml
deleted file mode 100644
index 4176428..0000000
--- a/docker-compose.yml
+++ /dev/null
@@ -1,23 +0,0 @@
-version: "3"
-
-services:
-  ovs:
-    image: libovsdb/ovs:latest
-    ports:
-      - "6640/tcp"
-    cap_add:
-      - NET_ADMIN
-    tty: true
-  test:
-    build:
-      context: .
-      dockerfile: Dockerfile.dev
-    depends_on:
-      - ovs
-    volumes: 
-      - .:/go/src/github.com/ovn-org/libovsdb
-    working_dir: /go/src/github.com/ovn-org/libovsdb
-    environment:
-      DOCKER_IP: "ovs"
-      OVS_DB: "tcp:ovs:6640"
-    command: make local integration-test-local
\ No newline at end of file
diff --git a/go.mod b/go.mod
index 56a7ffb..85ab368 100644
--- a/go.mod
+++ b/go.mod
@@ -7,5 +7,6 @@ require (
 	github.com/cenkalti/hub v1.0.1 // indirect
 	github.com/cenkalti/rpc2 v0.0.0-20210220005819-4a29bc83afe1
 	github.com/google/uuid v1.2.0
+	github.com/ory/dockertest/v3 v3.7.0
 	github.com/stretchr/testify v1.6.1
 )
diff --git a/go.sum b/go.sum
index e01ad44..231f21d 100644
--- a/go.sum
+++ b/go.sum
@@ -1,19 +1,126 @@
+github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78 h1:w+iIsaOQNcT7OZ575w+acHgRric5iCyQh+xv+KJ4HB8=
+github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78/go.mod h1:LmzpDX56iTiv29bbRTIsUNlaFfuhWRQBWjQdVyAevI8=
+github.com/Microsoft/go-winio v0.4.14 h1:+hMXMk01us9KgxGb7ftKQt2Xpf5hH/yky+TDA+qxleU=
+github.com/Microsoft/go-winio v0.4.14/go.mod h1:qXqCSQ3Xa7+6tgxaGTIe4Kpcdsi+P8jBhyzoq1bpyYA=
+github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5 h1:TngWCqHvy9oXAN6lEVMRuU21PR1EtLVZJmdB18Gu3Rw=
+github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5/go.mod h1:lmUJ/7eu/Q8D7ML55dXQrVaamCz2vxCfdQBasLZfHKk=
 github.com/cenk/hub v1.0.1 h1:RBwXNOF4a8KjD8BJ08XqN8KbrqaGiQLDrgvUGJSHuPA=
 github.com/cenk/hub v1.0.1/go.mod h1:rJM1LNAW0ppT8FMMuPK6c2NP/R2nH/UthtuRySSaf6Y=
+github.com/cenkalti/backoff/v4 v4.1.0 h1:c8LkOFQTzuO0WBM/ae5HdGQuZPfPxp7lqBRwQRm4fSc=
+github.com/cenkalti/backoff/v4 v4.1.0/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw=
 github.com/cenkalti/hub v1.0.1 h1:UMtjc6dHSaOQTO15SVA50MBIR9zQwvsukQupDrkIRtg=
 github.com/cenkalti/hub v1.0.1/go.mod h1:tcYwtS3a2d9NO/0xDXVJWx3IedurUjYCqFCmpi0lpHs=
 github.com/cenkalti/rpc2 v0.0.0-20210220005819-4a29bc83afe1 h1:aT9Ez2drLmrviqTnVnH87AeXLXLgUrXACJ2g90cTT2w=
 github.com/cenkalti/rpc2 v0.0.0-20210220005819-4a29bc83afe1/go.mod h1:v2npkhrXyk5BCnkNIiPdRI23Uq6uWPUQGL2hnRcRr/M=
-github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
+github.com/containerd/continuity v0.0.0-20190827140505-75bee3e2ccb6 h1:NmTXa/uVnDyp0TY5MKi197+3HWcnYWfnHGyaFthlnGw=
+github.com/containerd/continuity v0.0.0-20190827140505-75bee3e2ccb6/go.mod h1:GL3xCUCBDV3CZiTSEKksMWbLE66hEyuu9qyDOOqM47Y=
+github.com/creack/pty v1.1.11 h1:07n33Z8lZxZ2qwegKbObQohDhXDQxiMMz1NOUGYlesw=
+github.com/creack/pty v1.1.11/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/docker/cli v20.10.7+incompatible h1:pv/3NqibQKphWZiAskMzdz8w0PRbtTaEB+f6NwdU7Is=
+github.com/docker/cli v20.10.7+incompatible/go.mod h1:JLrzqnKDaYBop7H2jaqPtU4hHvMKP+vjCwu2uszcLI8=
+github.com/docker/docker v20.10.7+incompatible h1:Z6O9Nhsjv+ayUEeI1IojKbYcsGdgYSNqxe1s2MYzUhQ=
+github.com/docker/docker v20.10.7+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
+github.com/docker/go-connections v0.4.0 h1:El9xVISelRB7BuFusrZozjnkIM5YnzCViNKohAFqRJQ=
+github.com/docker/go-connections v0.4.0/go.mod h1:Gbd7IOopHjR8Iph03tsViu4nIes5XhDvyHbTtUxmeec=
+github.com/docker/go-units v0.4.0 h1:3uh0PgVws3nIA0Q+MwDC8yjEPf9zjRfZZWXZYDct3Tw=
+github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
+github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
+github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
+github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
+github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
+github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 h1:El6M4kTTCOh6aBiKaUGG7oYTSPP8MxqL4YI3kZKwcP4=
+github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510/go.mod h1:pupxD2MaaD3pAXIBCelhxNneeOaAeabZDe5s4K6zSpQ=
 github.com/google/uuid v1.2.0 h1:qJYtXnJRWmpe7m/3XlyhrsLrEURqHRM2kxzoxXqyUDs=
 github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/imdario/mergo v0.3.12 h1:b6R2BslTbIEToALKP7LxUvijTsNI9TAe80pLWN2g/HU=
+github.com/imdario/mergo v0.3.12/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
+github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
+github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
+github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
+github.com/konsorten/go-windows-terminal-sequences v1.0.2 h1:DB17ag19krx9CFsz4o3enTrPXyIXCl+2iCXH/aMAp9s=
+github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
+github.com/lib/pq v0.0.0-20180327071824-d34b9ff171c2 h1:hRGSmZu7j271trc9sneMrpOW7GN5ngLm8YUZIPzf394=
+github.com/lib/pq v0.0.0-20180327071824-d34b9ff171c2/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
+github.com/mitchellh/mapstructure v1.4.1 h1:CpVNEelQCZBooIPDn+AR3NpivK/TIKU8bDxdASFVQag=
+github.com/mitchellh/mapstructure v1.4.1/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
+github.com/moby/term v0.0.0-20201216013528-df9cb8a40635 h1:rzf0wL0CHVc8CEsgyygG0Mn9CNCCPZqOPaz8RiiHYQk=
+github.com/moby/term v0.0.0-20201216013528-df9cb8a40635/go.mod h1:FBS0z0QWA44HXygs7VXDUOGoN/1TV3RuWkLO04am3wc=
+github.com/opencontainers/go-digest v1.0.0-rc1 h1:WzifXhOVOEOuFYOJAW6aQqW0TooG2iki3E3Ii+WN7gQ=
+github.com/opencontainers/go-digest v1.0.0-rc1/go.mod h1:cMLVZDEM3+U2I4VmLI6N8jQYUd2OVphdqWwCJHrFt2s=
+github.com/opencontainers/image-spec v1.0.1 h1:JMemWkRwHx4Zj+fVxWoMCFm/8sYGGrUVojFA6h/TRcI=
+github.com/opencontainers/image-spec v1.0.1/go.mod h1:BtxoFyWECRxE4U/7sNtV5W15zMzWCbyJoFRP3s7yZA0=
+github.com/opencontainers/runc v1.0.0-rc9 h1:/k06BMULKF5hidyoZymkoDCzdJzltZpz/UU4LguQVtc=
+github.com/opencontainers/runc v1.0.0-rc9/go.mod h1:qT5XzbpPznkRYVz/mWwUaVBUv2rmF59PVA73FjuZG0U=
+github.com/ory/dockertest/v3 v3.7.0 h1:Bijzonc69Ont3OU0a3TWKJ1Rzlh3TsDXP1JrTAkSmsM=
+github.com/ory/dockertest/v3 v3.7.0/go.mod h1:PvCCgnP7AfBZeVrzwiUTjZx/IUXlGLC1zQlUQrLIlUE=
+github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/sirupsen/logrus v1.4.1/go.mod h1:ni0Sbl8bgC9z8RoU9G6nDWqqs/fq4eDPysMBDgk/93Q=
+github.com/sirupsen/logrus v1.4.2 h1:SPIRibHv4MatM3XXNO2BJeFLZwZ2LvZgfQ5+UNI2im4=
+github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
+github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f h1:J9EGpcZtP0E/raorCMxlFGSTBrsSlaDGf3jU/qvAE2c=
+github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
+github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 h1:EzJWgHovont7NscjpAxXsDA8S8BMYve8Y5+7cuRE7R0=
+github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415/go.mod h1:GwrjFmJcFw6At/Gs6z4yjiIwzuJ1/+UwLxMQDVQXShQ=
+github.com/xeipuuv/gojsonschema v1.2.0 h1:LhYJRs+L4fBtjZUfuSZIKGeVu0QRy8e5Xi7D17UxZ74=
+github.com/xeipuuv/gojsonschema v1.2.0/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQluxsYJ78Id3Y=
+github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20201021035429-f5854403a974 h1:IX6qOQeG5uLjB/hjjwjedwfjND0hgjPMMyO1RoIXQNI=
+golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190507160741-ecd444e8653b/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200831180312-196b9ba8737a/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f h1:+Nyd8tzPX9R7BWHguqsrbFdRx3WQ/1ib8I44HXV5yTA=
+golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20190624222133-a101b041ded4/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
+golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
+golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
+golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE=
+golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.3.0 h1:clyUAQHOM3G0M3f5vQj7LuJrETvjVot3Z5el9nffUtU=
+gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gotest.tools/v3 v3.0.2 h1:kG1BFyqVHuQoVQiR1bWGnfz/fmHvvuiSPIV7rvl360E=
+gotest.tools/v3 v3.0.2/go.mod h1:3SzNCllyD9/Y+b5r9JIKQ474KzkZyqLqEfYqMsX94Bk=
diff --git a/ovsdb/schema.go b/ovsdb/schema.go
index ec12202..ec7932d 100644
--- a/ovsdb/schema.go
+++ b/ovsdb/schema.go
@@ -64,33 +64,38 @@ func SchemaFromFile(f *os.File) (*DatabaseSchema, error) {
 // ValidateOperations performs basic validation for operations against a DatabaseSchema
 func (schema DatabaseSchema) ValidateOperations(operations ...Operation) bool {
 	for _, op := range operations {
-		table, ok := schema.Tables[op.Table]
-		if ok {
-			for column := range op.Row {
-				if _, ok := table.Columns[column]; !ok {
-					if column != "_uuid" && column != "_version" {
-						return false
-					}
-				}
-			}
-			for _, row := range op.Rows {
-				for column := range row {
+		switch op.Op {
+		case OperationAbort, OperationAssert, OperationComment, OperationCommit, OperationWait:
+			continue
+		case OperationInsert, OperationSelect, OperationUpdate, OperationMutate, OperationDelete:
+			table, ok := schema.Tables[op.Table]
+			if ok {
+				for column := range op.Row {
 					if _, ok := table.Columns[column]; !ok {
 						if column != "_uuid" && column != "_version" {
 							return false
 						}
 					}
 				}
-			}
-			for _, column := range op.Columns {
-				if _, ok := table.Columns[column]; !ok {
-					if column != "_uuid" && column != "_version" {
-						return false
+				for _, row := range op.Rows {
+					for column := range row {
+						if _, ok := table.Columns[column]; !ok {
+							if column != "_uuid" && column != "_version" {
+								return false
+							}
+						}
+					}
+				}
+				for _, column := range op.Columns {
+					if _, ok := table.Columns[column]; !ok {
+						if column != "_uuid" && column != "_version" {
+							return false
+						}
 					}
 				}
+			} else {
+				return false
 			}
-		} else {
-			return false
 		}
 	}
 	return true


Test patch:
diff --git a/client/client_test.go b/client/client_test.go
index db24c81..23967cd 100644
--- a/client/client_test.go
+++ b/client/client_test.go
@@ -6,6 +6,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/cenkalti/rpc2"
 	"github.com/ovn-org/libovsdb/cache"
 	"github.com/ovn-org/libovsdb/model"
 	"github.com/ovn-org/libovsdb/ovsdb"
@@ -85,6 +86,13 @@ type OpenvSwitch struct {
 	SystemVersion   []string          `ovsdb:"system_version"`
 }
 
+var defDB, _ = model.NewDBModel("Open_vSwitch",
+	map[string]model.Model{
+		"Open_vSwitch": &OpenvSwitch{},
+		"Bridge":       &Bridge{},
+	},
+)
+
 var schema = `{
 	"name": "Open_vSwitch",
 	"version": "8.2.0",
@@ -724,6 +732,11 @@ func TestUpdate(t *testing.T) {
 func TestOperationWhenNotConnected(t *testing.T) {
 	ovs, err := newOVSDBClient(defDB)
 	require.NoError(t, err)
+	var s ovsdb.DatabaseSchema
+	err = json.Unmarshal([]byte(schema), &s)
+	require.NoError(t, err)
+	ovs.schema = &s
+
 	tests := []struct {
 		name string
 		fn   func() error
@@ -737,18 +750,13 @@ func TestOperationWhenNotConnected(t *testing.T) {
 		{
 			"transact",
 			func() error {
-				_, err := ovs.Transact(ovsdb.Operation{})
+				comment := "this is only a test"
+				_, err := ovs.Transact(ovsdb.Operation{Op: ovsdb.OperationComment, Comment: &comment})
 				return err
 			},
 		},
 		{
-			"monitor",
-			func() error {
-				return ovs.Monitor("")
-			},
-		},
-		{
-			"monitor all",
+			"monitor/monitor all",
 			func() error {
 				return ovs.MonitorAll("")
 			},
@@ -776,7 +784,7 @@ func TestSetOption(t *testing.T) {
 	err = o.SetOption(WithEndpoint("tcp::6640"))
 	require.NoError(t, err)
 
-	o.connected = true
+	o.rpcClient = &rpc2.Client{}
 
 	err = o.SetOption(WithEndpoint("tcp::6641"))
 	assert.EqualError(t, err, "cannot set option when client is connected")
diff --git a/client/ovs_integration_test.go b/client/ovs_integration_test.go
deleted file mode 100644
index 1908746..0000000
--- a/client/ovs_integration_test.go
+++ /dev/null
@@ -1,547 +0,0 @@
-package client
-
-import (
-	"bytes"
-	"context"
-	"log"
-	"os"
-	"testing"
-	"time"
-
-	"github.com/ovn-org/libovsdb/model"
-	"github.com/ovn-org/libovsdb/ovsdb"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-const (
-	defOvsRunDir = "/var/run/openvswitch"
-	defOvsSocket = "db.sock"
-)
-
-// ORMBridge is the simplified ORM model of the Bridge table
-type bridgeType struct {
-	UUID        string            `ovsdb:"_uuid"`
-	Name        string            `ovsdb:"name"`
-	OtherConfig map[string]string `ovsdb:"other_config"`
-	ExternalIds map[string]string `ovsdb:"external_ids"`
-	Ports       []string          `ovsdb:"ports"`
-	Status      map[string]string `ovsdb:"status"`
-}
-
-// ORMovs is the simplified ORM model of the Bridge table
-type ovsType struct {
-	UUID    string   `ovsdb:"_uuid"`
-	Bridges []string `ovsdb:"bridges"`
-}
-
-var defDB, _ = model.NewDBModel("Open_vSwitch", map[string]model.Model{
-	"Open_vSwitch": &ovsType{},
-	"Bridge":       &bridgeType{}})
-
-var cfg *Config
-
-func SetConfig() {
-	cfg = &Config{}
-	var ovsRunDir = os.Getenv("OVS_RUNDIR")
-	if ovsRunDir == "" {
-		ovsRunDir = defOvsRunDir
-	}
-	var ovsDb = os.Getenv("OVS_DB")
-	if ovsDb == "" {
-		cfg.Addr = "unix:" + ovsRunDir + "/" + defOvsSocket
-	} else {
-		cfg.Addr = ovsDb
-	}
-}
-
-func TestConnectIntegration(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test")
-	}
-	SetConfig()
-	if testing.Short() {
-		t.Skip()
-	}
-	connected := make(chan bool)
-	errs := make(chan error)
-
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	defer cancel()
-
-	go func() {
-		ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
-		if err != nil {
-			errs <- err
-			return
-		}
-		err = ovs.Connect(ctx)
-		if err != nil {
-			errs <- err
-		} else {
-			connected <- true
-			ovs.Disconnect()
-		}
-	}()
-
-	select {
-	case err := <-errs:
-		t.Fatal(err)
-	case <-time.After(10 * time.Second):
-		t.Fatal("Connection Timed Out")
-	case <-connected:
-		return
-	}
-}
-
-func TestConnectReconnectIntegration(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test")
-	}
-	SetConfig()
-	if testing.Short() {
-		t.Skip()
-	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	defer cancel()
-
-	ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
-	require.NoError(t, err)
-	err = ovs.Connect(ctx)
-	require.NoError(t, err)
-
-	assert.Equal(t, true, ovs.Connected())
-
-	err = ovs.Echo()
-	require.NoError(t, err)
-
-	// make another connect call, this should return without error as we're already connected
-	ctx, cancel = context.WithTimeout(context.Background(), 2*time.Second)
-	defer cancel()
-	err = ovs.Connect(ctx)
-	require.NoError(t, err)
-
-	disconnectNotification := ovs.DisconnectNotify()
-	notified := make(chan struct{})
-	ready := make(chan struct{})
-
-	go func() {
-		ready <- struct{}{}
-		<-disconnectNotification
-		notified <- struct{}{}
-	}()
-
-	<-ready
-	ovs.Disconnect()
-
-	select {
-	case <-notified:
-		// got notification
-	case <-time.After(5 * time.Second):
-		t.Fatal("expected a disconnect notification but didn't receive one")
-	}
-
-	assert.Equal(t, false, ovs.Connected())
-
-	err = ovs.Echo()
-	require.EqualError(t, err, ErrNotConnected.Error())
-
-	ctx, cancel = context.WithTimeout(context.Background(), 10*time.Second)
-	defer cancel()
-
-	err = ovs.Connect(ctx)
-	require.NoError(t, err)
-	defer ovs.Disconnect()
-
-	err = ovs.Echo()
-	assert.NoError(t, err)
-
-}
-
-func TestListDbsIntegration(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test")
-	}
-	SetConfig()
-	if testing.Short() {
-		t.Skip()
-	}
-	ovs, err := newOVSDBClient(defDB, WithEndpoint(cfg.Addr))
-	require.NoError(t, err)
-	err = ovs.Connect(context.Background())
-	require.NoError(t, err)
-	reply, err := ovs.listDbs()
-	if err != nil {
-		log.Fatal("ListDbs error:", err)
-	}
-
-	found := false
-	for _, db := range reply {
-		if db == "Open_vSwitch" {
-			found = true
-		}
-	}
-
-	if !found {
-		t.Error("Expected: 'Open_vSwitch'", reply)
-	}
-	var b bytes.Buffer
-	ovs.Schema().Print(&b)
-	ovs.Disconnect()
-}
-
-func TestGetSchemasIntegration(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test")
-	}
-	SetConfig()
-	if testing.Short() {
-		t.Skip()
-	}
-
-	ovs, err := newOVSDBClient(defDB, WithEndpoint(cfg.Addr))
-	require.NoError(t, err)
-	err = ovs.Connect(context.Background())
-	require.NoError(t, err)
-	dbName := "Open_vSwitch"
-	reply, err := ovs.getSchema(dbName)
-
-	if err != nil {
-		log.Fatal("GetSchemas error:", err)
-		t.Error("Error Processing GetSchema for ", dbName, err)
-	}
-
-	if reply.Name != dbName {
-		t.Error("Schema Name mismatch. Expected: ", dbName, "Got: ", reply.Name)
-	}
-	ovs.Disconnect()
-}
-
-var bridgeName = "gopher-br7"
-var bridgeUUID string
-
-func TestInsertTransactIntegration(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test")
-	}
-	SetConfig()
-
-	ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
-	require.NoError(t, err)
-	err = ovs.Connect(context.Background())
-	require.NoError(t, err)
-	err = ovs.MonitorAll(nil)
-	assert.Nil(t, err)
-
-	// NamedUUID is used to add multiple related Operations in a single Transact operation
-	namedUUID := "gopher"
-	br := bridgeType{
-		UUID: namedUUID,
-		Name: bridgeName,
-		ExternalIds: map[string]string{
-			"go":     "awesome",
-			"docker": "made-for-each-other",
-		},
-	}
-
-	insertOp, err := ovs.Create(&br)
-	assert.Nil(t, err)
-
-	// Inserting a Bridge row in Bridge table requires mutating the open_vswitch table.
-	ovsRow := ovsType{}
-	mutateOp, err := ovs.WhereCache(func(*ovsType) bool { return true }).
-		Mutate(&ovsRow, model.Mutation{
-			Field:   &ovsRow.Bridges,
-			Mutator: ovsdb.MutateOperationInsert,
-			Value:   []string{namedUUID},
-		})
-	assert.Nil(t, err)
-
-	operations := append(insertOp, mutateOp...)
-	reply, err := ovs.Transact(operations...)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	operationErrs, err := ovsdb.CheckOperationResults(reply, operations)
-	if err != nil {
-		for _, oe := range operationErrs {
-			t.Error(oe)
-		}
-		t.Fatal(err)
-	}
-	bridgeUUID = reply[0].UUID.GoUUID
-	ovs.Disconnect()
-}
-
-func TestDeleteTransactIntegration(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test")
-	}
-	SetConfig()
-
-	if bridgeUUID == "" {
-		t.Skip()
-	}
-
-	ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
-	require.NoError(t, err)
-	err = ovs.Connect(context.Background())
-	require.NoError(t, err)
-	err = ovs.MonitorAll(nil)
-	assert.Nil(t, err)
-
-	deleteOp, err := ovs.Where(&bridgeType{Name: bridgeName}).Delete()
-	assert.Nil(t, err)
-
-	ovsRow := ovsType{}
-	mutateOp, err := ovs.WhereCache(func(*ovsType) bool { return true }).
-		Mutate(&ovsRow, model.Mutation{
-			Field:   &ovsRow.Bridges,
-			Mutator: ovsdb.MutateOperationDelete,
-			Value:   []string{bridgeUUID},
-		})
-
-	assert.Nil(t, err)
-
-	operations := append(deleteOp, mutateOp...)
-	reply, err := ovs.Transact(operations...)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	operationErrs, err := ovsdb.CheckOperationResults(reply, operations)
-	if err != nil {
-		for _, oe := range operationErrs {
-			t.Error(oe)
-		}
-		t.Fatal(err)
-	}
-	ovs.Disconnect()
-}
-
-func TestMonitorIntegration(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test")
-	}
-	SetConfig()
-	if testing.Short() {
-		t.Skip()
-	}
-
-	ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
-	require.NoError(t, err)
-	err = ovs.Connect(context.Background())
-	require.NoError(t, err)
-	err = ovs.MonitorAll(nil)
-	if err != nil {
-		t.Fatal(err)
-	}
-	ovs.Disconnect()
-}
-
-func TestTableSchemaValidationIntegration(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test")
-	}
-	SetConfig()
-
-	ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
-	require.NoError(t, err)
-	err = ovs.Connect(context.Background())
-	require.NoError(t, err)
-
-	operation := ovsdb.Operation{
-		Op:    "insert",
-		Table: "InvalidTable",
-		Row:   ovsdb.Row(map[string]interface{}{"name": "docker-ovs"}),
-	}
-	_, err = ovs.Transact(operation)
-
-	if err == nil {
-		t.Error("Invalid Table Name Validation failed")
-	}
-
-	ovs.Disconnect()
-}
-
-func TestColumnSchemaInRowValidationIntegration(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test")
-	}
-	SetConfig()
-
-	ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
-	require.NoError(t, err)
-	err = ovs.Connect(context.Background())
-	require.NoError(t, err)
-
-	operation := ovsdb.Operation{
-		Op:    "insert",
-		Table: "Bridge",
-		Row:   ovsdb.Row(map[string]interface{}{"name": "docker-ovs", "invalid_column": "invalid_column"}),
-	}
-
-	_, err = ovs.Transact(operation)
-
-	if err == nil {
-		t.Error("Invalid Column Name Validation failed")
-	}
-
-	ovs.Disconnect()
-}
-
-func TestColumnSchemaInMultipleRowsValidationIntegration(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test")
-	}
-	SetConfig()
-
-	ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
-	require.NoError(t, err)
-	err = ovs.Connect(context.Background())
-	require.NoError(t, err)
-
-	invalidBridge := ovsdb.Row(map[string]interface{}{"invalid_column": "invalid_column"})
-	bridge := ovsdb.Row(map[string]interface{}{"name": "docker-ovs"})
-	rows := []ovsdb.Row{invalidBridge, bridge}
-
-	operation := ovsdb.Operation{
-		Op:    "insert",
-		Table: "Bridge",
-		Rows:  rows,
-	}
-	_, err = ovs.Transact(operation)
-
-	if err == nil {
-		t.Error("Invalid Column Name Validation failed")
-	}
-
-	ovs.Disconnect()
-}
-
-func TestColumnSchemaValidationIntegration(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test")
-	}
-	SetConfig()
-	if testing.Short() {
-		t.Skip()
-	}
-
-	ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
-	require.NoError(t, err)
-	err = ovs.Connect(context.Background())
-	require.NoError(t, err)
-
-	operation := ovsdb.Operation{
-		Op:      "select",
-		Table:   "Bridge",
-		Columns: []string{"name", "invalidColumn"},
-	}
-	_, err = ovs.Transact(operation)
-
-	if err == nil {
-		t.Error("Invalid Column Name Validation failed")
-	}
-
-	ovs.Disconnect()
-}
-
-func TestMonitorCancelIntegration(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test")
-	}
-	SetConfig()
-
-	ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
-	require.NoError(t, err)
-	err = ovs.Connect(context.Background())
-	require.NoError(t, err)
-
-	monitorID := "f1b2ca48-aad7-11e7-abc4-cec278b6b50a"
-
-	requests := make(map[string]ovsdb.MonitorRequest)
-	requests["Bridge"] = ovsdb.MonitorRequest{
-		Columns: []string{"name"},
-		Select:  ovsdb.NewDefaultMonitorSelect(),
-	}
-
-	err = ovs.Monitor(monitorID,
-		ovs.NewTableMonitor(&ovsType{}),
-		ovs.NewTableMonitor(&bridgeType{}),
-	)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	err = ovs.MonitorCancel(monitorID)
-
-	if err != nil {
-		t.Error("MonitorCancel operation failed with error=", err)
-	}
-	ovs.Disconnect()
-}
-
-func TestInsertDuplicateTransactIntegration(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test")
-	}
-	SetConfig()
-
-	ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
-	require.NoError(t, err)
-	err = ovs.Connect(context.Background())
-	require.NoError(t, err)
-
-	err = ovs.MonitorAll(nil)
-	assert.Nil(t, err)
-
-	// NamedUUID is used to add multiple related Operations in a single Transact operation
-	namedUUID := "gopher"
-	br := bridgeType{
-		UUID: namedUUID,
-		Name: "br-dup",
-		ExternalIds: map[string]string{
-			"go":     "awesome",
-			"docker": "made-for-each-other",
-		},
-	}
-
-	insertOp, err := ovs.Create(&br)
-	assert.Nil(t, err)
-
-	// Inserting a Bridge row in Bridge table requires mutating the open_vswitch table.
-	ovsRow := ovsType{}
-	mutateOp, err := ovs.WhereCache(func(*ovsType) bool { return true }).
-		Mutate(&ovsRow, model.Mutation{
-			Field:   &ovsRow.Bridges,
-			Mutator: ovsdb.MutateOperationInsert,
-			Value:   []string{namedUUID},
-		})
-	assert.Nil(t, err)
-
-	operations := append(insertOp, mutateOp...)
-	reply, err := ovs.Transact(operations...)
-	if err != nil {
-		t.Fatal(err)
-	}
-	operationErrs, err := ovsdb.CheckOperationResults(reply, operations)
-	if err != nil {
-		for _, oe := range operationErrs {
-			t.Error(oe)
-		}
-		t.Fatal(err)
-	}
-	bridgeUUID = reply[0].UUID.GoUUID
-
-	reply, err = ovs.Transact(operations...)
-	if err != nil {
-		t.Fatal(err)
-	}
-	_, err = ovsdb.CheckOperationResults(reply, operations)
-	assert.Error(t, err)
-	assert.IsType(t, &ovsdb.ConstraintViolation{}, err)
-	ovs.Disconnect()
-}
diff --git a/test/ovs/ovs_integration_test.go b/test/ovs/ovs_integration_test.go
new file mode 100644
index 0000000..34066bb
--- /dev/null
+++ b/test/ovs/ovs_integration_test.go
@@ -0,0 +1,379 @@
+package ovs
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/ory/dockertest/v3"
+	"github.com/ory/dockertest/v3/docker"
+	"github.com/ovn-org/libovsdb/client"
+	"github.com/ovn-org/libovsdb/model"
+	"github.com/ovn-org/libovsdb/ovsdb"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
+)
+
+// OVSIntegrationSuite runs tests against a real Open vSwitch instance
+type OVSIntegrationSuite struct {
+	suite.Suite
+	pool     *dockertest.Pool
+	resource *dockertest.Resource
+	client   client.Client
+}
+
+func (suite *OVSIntegrationSuite) SetupSuite() {
+	var err error
+	suite.pool, err = dockertest.NewPool("")
+	require.NoError(suite.T(), err)
+
+	tag := os.Getenv("OVS_IMAGE_TAG")
+	if tag == "" {
+		tag = "latest"
+	}
+
+	options := &dockertest.RunOptions{
+		Repository:   "libovsdb/ovs",
+		Tag:          tag,
+		ExposedPorts: []string{"6640/tcp"},
+		Tty:          true,
+	}
+	hostConfig := func(config *docker.HostConfig) {
+		// set AutoRemove to true so that stopped container goes away by itself
+		config.AutoRemove = true
+		config.RestartPolicy = docker.RestartPolicy{
+			Name: "no",
+		}
+	}
+
+	suite.resource, err = suite.pool.RunWithOptions(options, hostConfig)
+	require.NoError(suite.T(), err)
+
+	// set expiry to 30 seconds so containers are cleaned up on test panic
+	err = suite.resource.Expire(30)
+	require.NoError(suite.T(), err)
+
+	// let the container start before we attempt connection
+	time.Sleep(5 * time.Second)
+
+	err = suite.pool.Retry(func() error {
+		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+		defer cancel()
+		endpoint := fmt.Sprintf("tcp::%s", suite.resource.GetPort("6640/tcp"))
+		ovs, err := client.NewOVSDBClient(defDB, client.WithEndpoint(endpoint))
+		if err != nil {
+			return err
+		}
+		err = ovs.Connect(ctx)
+		if err != nil {
+			return err
+		}
+		suite.client = ovs
+		return nil
+	})
+	require.NoError(suite.T(), err)
+
+	// give ovsdb-server some time to start up
+
+	err = suite.client.MonitorAll(nil)
+	require.NoError(suite.T(), err)
+}
+
+func (suite *OVSIntegrationSuite) TearDownSuite() {
+	if suite.client != nil {
+		suite.client.Disconnect()
+		suite.client = nil
+	}
+	err := suite.pool.Purge(suite.resource)
+	require.NoError(suite.T(), err)
+}
+
+func TestOVSIntegrationTestSuite(t *testing.T) {
+	if testing.Short() {
+		t.Skip()
+	}
+	suite.Run(t, new(OVSIntegrationSuite))
+}
+
+// bridgeType is the simplified ORM model of the Bridge table
+type bridgeType struct {
+	UUID        string            `ovsdb:"_uuid"`
+	Name        string            `ovsdb:"name"`
+	OtherConfig map[string]string `ovsdb:"other_config"`
+	ExternalIds map[string]string `ovsdb:"external_ids"`
+	Ports       []string          `ovsdb:"ports"`
+	Status      map[string]string `ovsdb:"status"`
+}
+
+// ovsType is the simplified ORM model of the Bridge table
+type ovsType struct {
+	UUID    string   `ovsdb:"_uuid"`
+	Bridges []string `ovsdb:"bridges"`
+}
+
+var defDB, _ = model.NewDBModel("Open_vSwitch", map[string]model.Model{
+	"Open_vSwitch": &ovsType{},
+	"Bridge":       &bridgeType{}})
+
+func (suite *OVSIntegrationSuite) TestConnectReconnectIntegration() {
+	assert.True(suite.T(), suite.client.Connected())
+	err := suite.client.Echo()
+	require.NoError(suite.T(), err)
+
+	// make another connect call, this should return without error as we're already connected
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	err = suite.client.Connect(ctx)
+	require.NoError(suite.T(), err)
+
+	disconnectNotification := suite.client.DisconnectNotify()
+	notified := make(chan struct{})
+	ready := make(chan struct{})
+
+	go func() {
+		ready <- struct{}{}
+		<-disconnectNotification
+		notified <- struct{}{}
+	}()
+
+	<-ready
+	suite.client.Disconnect()
+
+	select {
+	case <-notified:
+		// got notification
+	case <-time.After(5 * time.Second):
+		suite.T().Fatal("expected a disconnect notification but didn't receive one")
+	}
+
+	assert.Equal(suite.T(), false, suite.client.Connected())
+
+	err = suite.client.Echo()
+	require.EqualError(suite.T(), err, client.ErrNotConnected.Error())
+
+	ctx, cancel = context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	err = suite.client.Connect(ctx)
+	require.NoError(suite.T(), err)
+
+	err = suite.client.Echo()
+	assert.NoError(suite.T(), err)
+
+	err = suite.client.MonitorAll(nil)
+	require.NoError(suite.T(), err)
+}
+func (suite *OVSIntegrationSuite) TestInsertTransactIntegration() {
+	bridgeName := "gopher-br7"
+
+	// NamedUUID is used to add multiple related Operations in a single Transact operation
+	namedUUID := "gopher"
+	br := bridgeType{
+		UUID: namedUUID,
+		Name: bridgeName,
+		ExternalIds: map[string]string{
+			"go":     "awesome",
+			"docker": "made-for-each-other",
+		},
+	}
+
+	insertOp, err := suite.client.Create(&br)
+	require.NoError(suite.T(), err)
+
+	// Inserting a Bridge row in Bridge table requires mutating the open_vswitch table.
+	ovsRow := ovsType{}
+	mutateOp, err := suite.client.WhereCache(func(*ovsType) bool { return true }).
+		Mutate(&ovsRow, model.Mutation{
+			Field:   &ovsRow.Bridges,
+			Mutator: ovsdb.MutateOperationInsert,
+			Value:   []string{namedUUID},
+		})
+	require.NoError(suite.T(), err)
+
+	operations := append(insertOp, mutateOp...)
+	reply, err := suite.client.Transact(operations...)
+	require.NoError(suite.T(), err)
+
+	operationErrs, err := ovsdb.CheckOperationResults(reply, operations)
+	if err != nil {
+		for _, oe := range operationErrs {
+			suite.T().Error(oe)
+		}
+		suite.T().Fatal(err)
+	}
+}
+
+func (suite *OVSIntegrationSuite) TestInsertAndDeleteTransactIntegration() {
+	bridgeName := "gopher-br5"
+	namedUUID := "gopher"
+	br := bridgeType{
+		UUID: namedUUID,
+		Name: bridgeName,
+		ExternalIds: map[string]string{
+			"go":     "awesome",
+			"docker": "made-for-each-other",
+		},
+	}
+
+	insertOp, err := suite.client.Create(&br)
+	require.NoError(suite.T(), err)
+
+	// Inserting a Bridge row in Bridge table requires mutating the open_vswitch table.
+	ovsRow := ovsType{}
+	mutateOp, err := suite.client.WhereCache(func(*ovsType) bool { return true }).
+		Mutate(&ovsRow, model.Mutation{
+			Field:   &ovsRow.Bridges,
+			Mutator: ovsdb.MutateOperationInsert,
+			Value:   []string{namedUUID},
+		})
+	require.NoError(suite.T(), err)
+
+	operations := append(insertOp, mutateOp...)
+	reply, err := suite.client.Transact(operations...)
+	require.NoError(suite.T(), err)
+
+	operationErrs, err := ovsdb.CheckOperationResults(reply, operations)
+	if err != nil {
+		for _, oe := range operationErrs {
+			suite.T().Error(oe)
+		}
+		suite.T().Fatal(err)
+	}
+	bridgeUUID := reply[0].UUID.GoUUID
+
+	deleteOp, err := suite.client.Where(&bridgeType{Name: bridgeName}).Delete()
+	require.NoError(suite.T(), err)
+
+	delMutateOp, err := suite.client.WhereCache(func(*ovsType) bool { return true }).
+		Mutate(&ovsRow, model.Mutation{
+			Field:   &ovsRow.Bridges,
+			Mutator: ovsdb.MutateOperationDelete,
+			Value:   []string{bridgeUUID},
+		})
+
+	require.NoError(suite.T(), err)
+
+	delOperations := append(deleteOp, delMutateOp...)
+	delReply, err := suite.client.Transact(delOperations...)
+	require.NoError(suite.T(), err)
+
+	delOperationErrs, err := ovsdb.CheckOperationResults(delReply, delOperations)
+	if err != nil {
+		for _, oe := range delOperationErrs {
+			suite.T().Error(oe)
+		}
+		suite.T().Fatal(err)
+	}
+}
+
+func (suite *OVSIntegrationSuite) TestTableSchemaValidationIntegration() {
+	operation := ovsdb.Operation{
+		Op:    "insert",
+		Table: "InvalidTable",
+		Row:   ovsdb.Row(map[string]interface{}{"name": "docker-ovs"}),
+	}
+	_, err := suite.client.Transact(operation)
+	assert.Error(suite.T(), err)
+}
+
+func (suite *OVSIntegrationSuite) TestColumnSchemaInRowValidationIntegration() {
+	operation := ovsdb.Operation{
+		Op:    "insert",
+		Table: "Bridge",
+		Row:   ovsdb.Row(map[string]interface{}{"name": "docker-ovs", "invalid_column": "invalid_column"}),
+	}
+
+	_, err := suite.client.Transact(operation)
+	assert.Error(suite.T(), err)
+}
+
+func (suite *OVSIntegrationSuite) TestColumnSchemaInMultipleRowsValidationIntegration() {
+	invalidBridge := ovsdb.Row(map[string]interface{}{"invalid_column": "invalid_column"})
+	bridge := ovsdb.Row(map[string]interface{}{"name": "docker-ovs"})
+	rows := []ovsdb.Row{invalidBridge, bridge}
+
+	operation := ovsdb.Operation{
+		Op:    "insert",
+		Table: "Bridge",
+		Rows:  rows,
+	}
+	_, err := suite.client.Transact(operation)
+	assert.Error(suite.T(), err)
+}
+
+func (suite *OVSIntegrationSuite) TestColumnSchemaValidationIntegration() {
+	operation := ovsdb.Operation{
+		Op:      "select",
+		Table:   "Bridge",
+		Columns: []string{"name", "invalidColumn"},
+	}
+	_, err := suite.client.Transact(operation)
+	assert.Error(suite.T(), err)
+}
+
+func (suite *OVSIntegrationSuite) TestMonitorCancelIntegration() {
+	monitorID := "f1b2ca48-aad7-11e7-abc4-cec278b6b50a"
+
+	requests := make(map[string]ovsdb.MonitorRequest)
+	requests["Bridge"] = ovsdb.MonitorRequest{
+		Columns: []string{"name"},
+		Select:  ovsdb.NewDefaultMonitorSelect(),
+	}
+
+	err := suite.client.Monitor(monitorID,
+		suite.client.NewTableMonitor(&ovsType{}),
+		suite.client.NewTableMonitor(&bridgeType{}),
+	)
+	require.NoError(suite.T(), err)
+
+	err = suite.client.MonitorCancel(monitorID)
+	assert.NoError(suite.T(), err)
+}
+
+func (suite *OVSIntegrationSuite) TestInsertDuplicateTransactIntegration() {
+	// NamedUUID is used to add multiple related Operations in a single Transact operation
+	namedUUID := "gopher"
+	br := bridgeType{
+		UUID: namedUUID,
+		Name: "br-dup",
+		ExternalIds: map[string]string{
+			"go":     "awesome",
+			"docker": "made-for-each-other",
+		},
+	}
+
+	insertOp, err := suite.client.Create(&br)
+	require.NoError(suite.T(), err)
+
+	// Inserting a Bridge row in Bridge table requires mutating the open_vswitch table.
+	ovsRow := ovsType{}
+	mutateOp, err := suite.client.WhereCache(func(*ovsType) bool { return true }).
+		Mutate(&ovsRow, model.Mutation{
+			Field:   &ovsRow.Bridges,
+			Mutator: ovsdb.MutateOperationInsert,
+			Value:   []string{namedUUID},
+		})
+	require.NoError(suite.T(), err)
+
+	operations := append(insertOp, mutateOp...)
+	reply, err := suite.client.Transact(operations...)
+	require.NoError(suite.T(), err)
+
+	operationErrs, err := ovsdb.CheckOperationResults(reply, operations)
+	if err != nil {
+		for _, oe := range operationErrs {
+			suite.T().Error(oe)
+		}
+		suite.T().Fatal(err)
+	}
+
+	reply2, err := suite.client.Transact(operations...)
+	require.NoError(suite.T(), err)
+
+	_, err = ovsdb.CheckOperationResults(reply2, operations)
+	assert.Error(suite.T(), err)
+	assert.IsType(suite.T(), &ovsdb.ConstraintViolation{}, err)
+}