system-tests: add debug flag

snapshots come in 3 kinds: local, remote and snapshot

find resources to be locked for each kind, recursively for the snapshot kind

Makefile

@@ -29,6 +29,11 @@ ifeq ($(CAPTURE),1)
 CAPTURE_ARG := --capture
 endif
 
+# export DEBUG=1 to enable debug output in system tests
+ifeq ($(DEBUG),1)
+DEBUG_ARG := --debug
+endif
+
 help:  ## Print this help
 	@grep -E '^[a-zA-Z][a-zA-Z0-9_-]*:.*?## .*$$' $(MAKEFILE_LIST) | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-30s\033[0m %s\n", $$1, $$2}'
 
@@ -121,7 +126,7 @@ system-test: prepare swagger etcd-install  ## Run system tests
 	if [ ! -e ~/aptly-fixture-pool ]; then git clone https://github.com/aptly-dev/aptly-fixture-pool.git ~/aptly-fixture-pool/; fi
 	test -f ~/etcd.db || (curl -o ~/etcd.db.xz http://repo.aptly.info/system-tests/etcd.db.xz && xz -d ~/etcd.db.xz)
 	# Run system tests
-	PATH=$(BINPATH)/:$(PATH) FORCE_COLOR=1 $(PYTHON) system/run.py --long $(COVERAGE_ARG_TEST) $(CAPTURE_ARG) $(TEST)
+	PATH=$(BINPATH)/:$(PATH) FORCE_COLOR=1 $(PYTHON) system/run.py --long $(COVERAGE_ARG_TEST) $(CAPTURE_ARG) $(DEBUG_ARG) $(TEST)
 
 bench:
 	@echo "\e[33m\e[1mRunning benchmark ...\e[0m"
@@ -211,7 +216,7 @@ docker-system-test:  ## Run system tests in docker container (add TEST=t04_mirro
 		AZURE_STORAGE_ACCESS_KEY="Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==" \
 		AWS_ACCESS_KEY_ID=$(AWS_ACCESS_KEY_ID) \
 		AWS_SECRET_ACCESS_KEY=$(AWS_SECRET_ACCESS_KEY) \
-		system-test TEST=$(TEST) CAPTURE=$(CAPTURE) COVERAGE_SKIP=$(COVERAGE_SKIP) \
+		system-test TEST=$(TEST) CAPTURE=$(CAPTURE) COVERAGE_SKIP=$(COVERAGE_SKIP) DEBUG=$(DEBUG) \
 		azurite-stop
 
 docker-serve:  ## Run development server (auto recompiling) on http://localhost:3142

api/publish.go

@@ -255,13 +255,13 @@ func apiPublishRepoOrSnapshot(c *gin.Context) {
 	if b.SourceKind == deb.SourceSnapshot {
 		var snapshot *deb.Snapshot
 
-		snapshotCollection := collectionFactory.SnapshotCollection()
+		tmpCollection := collectionFactory.SnapshotCollection()
 
 		for _, source := range b.Sources {
 			components = append(components, source.Component)
 			names = append(names, source.Name)
 
-			snapshot, err = snapshotCollection.ByName(source.Name)
+			snapshot, err = tmpCollection.ByName(source.Name)
 			if err != nil {
 				AbortWithJSONError(c, http.StatusNotFound, fmt.Errorf("unable to publish: %s", err))
 				return
@@ -273,13 +273,13 @@ func apiPublishRepoOrSnapshot(c *gin.Context) {
 	} else if b.SourceKind == deb.SourceLocalRepo {
 		var localRepo *deb.LocalRepo
 
-		localCollection := collectionFactory.LocalRepoCollection()
+		tmpCollection := collectionFactory.LocalRepoCollection()
 
 		for _, source := range b.Sources {
 			components = append(components, source.Component)
 			names = append(names, source.Name)
 
-			localRepo, err = localCollection.ByName(source.Name)
+			localRepo, err = tmpCollection.ByName(source.Name)
 			if err != nil {
 				AbortWithJSONError(c, http.StatusNotFound, fmt.Errorf("unable to publish: %s", err))
 				return
@@ -332,8 +332,6 @@ func apiPublishRepoOrSnapshot(c *gin.Context) {
 			return &task.ProcessReturnValue{Code: http.StatusInternalServerError, Value: nil}, fmt.Errorf("unable to publish: %s", err)
 		}
 
-		resources = append(resources, string(published.Key()))
-
 		if b.Origin != "" {
 			published.Origin = b.Origin
 		}
@@ -387,6 +385,46 @@ func apiPublishRepoOrSnapshot(c *gin.Context) {
 	})
 }
 
+// Return resources to be locked for a Snapshot name
+func getSnapshotResources(snapshotCollection *deb.SnapshotCollection, snapshotName string) (resources []string, err error) {
+	snapshot, err := snapshotCollection.ByName(snapshotName)
+	if err != nil {
+		return
+	}
+	resources = append(resources, string(snapshot.ResourceKey()))
+
+	for _, sourceID := range snapshot.SourceIDs {
+		if snapshot.SourceKind == deb.SourceSnapshot {
+			snapshot2, err2 := snapshotCollection.ByUUID(sourceID)
+			if err2 != nil {
+				err = err2
+				return
+			}
+			res, err3 := getSnapshotResources(snapshotCollection, snapshot2.Name)
+			if err3 != nil {
+				err = err3
+				return
+			}
+			resources = append(resources, res...)
+		} else if snapshot.SourceKind == deb.SourceLocalRepo {
+			var repo *deb.LocalRepo
+			repo, err = context.NewCollectionFactory().LocalRepoCollection().ByUUID(sourceID)
+			if err != nil {
+				return
+			}
+			resources = append(resources, string(repo.Key()))
+		} else if snapshot.SourceKind == deb.SourceRemoteRepo {
+			var mirror *deb.RemoteRepo
+			mirror, err = context.NewCollectionFactory().RemoteRepoCollection().ByUUID(sourceID)
+			if err != nil {
+				return
+			}
+			resources = append(resources, string(mirror.Key()))
+		}
+	}
+	return
+}
+
 type publishedRepoUpdateSwitchParams struct {
 	// when publishing, overwrite files in pool/ directory without notice
 	ForceOverwrite bool `                         json:"ForceOverwrite" example:"false"`
@@ -406,12 +444,12 @@ type publishedRepoUpdateSwitchParams struct {
 	SignedBy *string `                            json:"SignedBy"  example:""`
 	// Enable multiple packages with the same filename in different distributions
 	MultiDist *bool `                             json:"MultiDist"      example:"false"`
-    // Value of Label: field in published repository stanza
+	// Value of Label: field in published repository stanza
-    Label *string `                               json:"Label"          example:"Debian"`
+	Label *string `                               json:"Label"          example:"Debian"`
-    // Value of Origin: field in published repository stanza
+	// Value of Origin: field in published repository stanza
-    Origin *string `                              json:"Origin"         example:"Debian"`
+	Origin *string `                              json:"Origin"         example:"Debian"`
-    // Version of the release: Optional
+	// Version of the release: Optional
-    Version *string `                             json:"Version"        example:"13.3"`
+	Version *string `                             json:"Version"        example:"13.3"`
 }
 
 // @Summary Update Published Repository
@@ -465,18 +503,31 @@ func apiPublishUpdateSwitch(c *gin.Context) {
 		return
 	}
 
+	resources := []string{string(published.Key())}
+
 	if published.SourceKind == deb.SourceLocalRepo {
 		if len(b.Snapshots) > 0 {
 			AbortWithJSONError(c, http.StatusBadRequest, fmt.Errorf("snapshots shouldn't be given when updating local repo"))
 			return
 		}
-	} else if published.SourceKind == deb.SourceSnapshot {
+
-		for _, snapshotInfo := range b.Snapshots {
+		localCollection := collectionFactory.LocalRepoCollection()
-			_, err2 := snapshotCollection.ByName(snapshotInfo.Name)
+		for _, sourceID := range published.Sources {
+			localRepo, err2 := localCollection.ByUUID(sourceID)
 			if err2 != nil {
 				AbortWithJSONError(c, http.StatusNotFound, err2)
 				return
 			}
+			resources = append(resources, string(localRepo.Key()))
+		}
+	} else if published.SourceKind == deb.SourceSnapshot {
+		for _, snapshotInfo := range b.Snapshots {
+			res, err2 := getSnapshotResources(snapshotCollection, snapshotInfo.Name)
+			if err2 != nil {
+				AbortWithJSONError(c, http.StatusNotFound, err2)
+				return
+			}
+			resources = append(resources, res...)
 		}
 	} else {
 		AbortWithJSONError(c, http.StatusInternalServerError, fmt.Errorf("unknown published repository type"))
@@ -515,7 +566,6 @@ func apiPublishUpdateSwitch(c *gin.Context) {
 		published.Version = *b.Version
 	}
 
-	resources := []string{string(published.Key())}
 	taskName := fmt.Sprintf("Update published %s repository %s/%s", published.SourceKind, published.StoragePrefix(), published.Distribution)
 	maybeRunTaskInBackground(c, taskName, resources, func(out aptly.Progress, _ *task.Detail) (*task.ProcessReturnValue, error) {
 		err = collection.LoadComplete(published, collectionFactory)

deb/publish.go

@@ -609,7 +609,12 @@ func (p *PublishedRepo) StoragePrefix() string {
 
 // Key returns unique key identifying PublishedRepo
 func (p *PublishedRepo) Key() []byte {
-	return []byte("U" + p.StoragePrefix() + ">>" + p.Distribution)
+	if p.MultiDist {
+		// do not lock Distribution in MultiDist
+		return []byte("UM" + p.StoragePrefix())
+	} else {
+		return []byte("U" + p.StoragePrefix() + ">>" + p.Distribution)
+	}
 }
 
 // RefKey is a unique id for package reference list

system/api_lib.py

@@ -25,7 +25,7 @@ class APITest(BaseTest):
     """
     aptly_server = None
     aptly_out = None
-    debugOutput = True
+    debugOutput = False  # Controlled by --debug flag in run.py
     base_url = "127.0.0.1:8765"
     configOverride = {
         "FileSystemPublishEndpoints": {

system/lib.py

@@ -164,10 +164,10 @@ class BaseTest(object):
             self.run()
             self.check()
         except Exception as exc:
-            if self.debugOutput:
-                print(f"API log:\n{self.debug_output()}")
             raise exc
         finally:
+            if self.debugOutput:
+                print(f"API log:\n{self.debug_output()}")
             self.teardown()
 
     def prepare_remove_all(self):

system/run.py

@@ -36,7 +36,7 @@ def natural_key(string_):
     return [int(s) if s.isdigit() else s for s in re.split(r'(\d+)', string_)]
 
 
-def run(include_long_tests=False, capture_results=False, tests=None, filters=None, coverage_dir=None, coverage_skip=False):
+def run(include_long_tests=False, capture_results=False, tests=None, filters=None, coverage_dir=None, coverage_skip=False, debug=False):
     """
     Run system test.
     """
@@ -50,6 +50,9 @@ def run(include_long_tests=False, capture_results=False, tests=None, filters=Non
     if not coverage_dir and not coverage_skip:
         coverage_dir = mkdtemp(suffix="aptly-coverage")
 
+    # Set debug output globally for all test classes
+    BaseTest.debugOutput = debug
+
     failed = False
     for test in tests:
         orig_stdout = sys.stdout
@@ -155,6 +158,7 @@ def run(include_long_tests=False, capture_results=False, tests=None, filters=Non
                     traceback.print_exception(typ, val, tb, file=orig_stdout)
                 else:
                     orig_stdout.write(colored("\b\b\b\bOK", color="green", attrs=["bold"]) + f"   {duration}\n")
+                    orig_stdout.write(testout.get_contents())
 
                 t.shutdown()
 
@@ -214,6 +218,7 @@ if __name__ == "__main__":
     capture_results = False
     coverage_dir = None
     coverage_skip = False
+    debug = False
     tests = None
     args = sys.argv[1:]
 
@@ -227,6 +232,8 @@ if __name__ == "__main__":
             args = args[1:]
         elif args[0] == "--coverage-skip":
             coverage_skip = True
+        elif args[0] == "--debug":
+            debug = True
 
         args = args[1:]
 
@@ -239,4 +246,4 @@ if __name__ == "__main__":
         else:
             filters.append(arg)
 
-    run(include_long_tests, capture_results, tests, filters, coverage_dir, coverage_skip)
+    run(include_long_tests, capture_results, tests, filters, coverage_dir, coverage_skip, debug)

system/t12_api/publish.py

@@ -992,6 +992,232 @@ class PublishSwitchAPITestRepo(APITest):
         self.check_not_exists("public/" + prefix + "dists/")
 
 
+class PublishSwitchAPITestMirror(APITest):
+    """
+    PUT /publish/:prefix/:distribution (snapshots), DELETE /publish/:prefix/:distribution
+    """
+    fixtureGpg = True
+
+    def check(self):
+        mirror_name = self.random_name()
+        mirror_desc = {'Name': mirror_name,
+                       'ArchiveURL': 'http://repo.aptly.info/system-tests/packagecloud.io/varnishcache/varnish30/debian/',
+                       'Distribution': 'wheezy',
+                       'Keyrings': ["aptlytest.gpg"],
+                       'Architectures': ["amd64"],
+                       'Components': ['main']}
+        mirror_desc['IgnoreSignatures'] = True
+
+        # Create Mirror
+        resp = self.post("/api/mirrors", json=mirror_desc)
+        self.check_equal(resp.status_code, 201)
+
+        # Get Mirror
+        resp = self.get("/api/mirrors/" + mirror_name + "/packages")
+        self.check_equal(resp.status_code, 404)
+
+        # Update Mirror
+        resp = self.put_task("/api/mirrors/" + mirror_name, json=mirror_desc)
+        self.check_task(resp)
+
+        # Snapshot Mirror
+        snapshot1_name = self.random_name()
+        task = self.post_task("/api/mirrors/" + mirror_name + '/snapshots', json={'Name': snapshot1_name})
+        self.check_task(task)
+
+        # Publish Snapshot
+        prefix = self.random_name()
+        task = self.post_task(
+            "/api/publish/" + prefix,
+            json={
+                "Architectures": ["i386", "source"],
+                "SourceKind": "snapshot",
+                "Sources": [{"Name": snapshot1_name}],
+                "Signing": DefaultSigningOptions,
+            })
+        self.check_task(task)
+
+        repo_expected = {
+            'AcquireByHash': False,
+            'Architectures': ['i386', 'source'],
+            'Codename': '',
+            'Distribution': 'wheezy',
+            'Label': '',
+            'NotAutomatic': '',
+            'ButAutomaticUpgrades': '',
+            'Origin': 'packagecloud.io/varnishcache/varnish30',
+            'Version': '',
+            'Path': prefix + '/' + 'wheezy',
+            'Prefix': prefix,
+            'SignedBy': '',
+            'SkipContents': False,
+            'MultiDist': False,
+            'SourceKind': 'snapshot',
+            'Sources': [{'Component': 'main', 'Name': snapshot1_name}],
+            'Storage': '',
+            'Suite': ''}
+        all_repos = self.get("/api/publish")
+        self.check_equal(all_repos.status_code, 200)
+        self.check_in(repo_expected, all_repos.json())
+
+        # Snapshot Mirror 2
+        snapshot2_name = self.random_name()
+        task = self.post_task("/api/mirrors/" + mirror_name + '/snapshots', json={'Name': snapshot2_name})
+        self.check_task(task)
+
+        task = self.put_task(
+            "/api/publish/" + prefix + "/wheezy",
+            json={
+                "Snapshots": [{"Component": "main", "Name": snapshot2_name}],
+                "Signing": DefaultSigningOptions,
+                "SkipContents": True,
+                "Label": "fun",
+                "Origin": "earth",
+                "Version": "13.3",
+            })
+        self.check_task(task)
+        repo_expected = {
+            'AcquireByHash': False,
+            'Architectures': ['i386', 'source'],
+            'Codename': '',
+            'Distribution': 'wheezy',
+            'Label': 'fun',
+            'Origin': 'earth',
+            'Version': '13.3',
+            'NotAutomatic': '',
+            'ButAutomaticUpgrades': '',
+            'Path': prefix + '/' + 'wheezy',
+            'Prefix': prefix,
+            'SignedBy': '',
+            'SkipContents': True,
+            'MultiDist': False,
+            'SourceKind': 'snapshot',
+            'Sources': [{'Component': 'main', 'Name': snapshot2_name}],
+            'Storage': '',
+            'Suite': ''}
+
+        all_repos = self.get("/api/publish")
+        self.check_equal(all_repos.status_code, 200)
+        self.check_in(repo_expected, all_repos.json())
+
+        task = self.delete_task("/api/publish/" + prefix + "/wheezy")
+        self.check_task(task)
+        self.check_not_exists("public/" + prefix + "dists/")
+
+
+class PublishSwitchAPITestSnapshot(APITest):
+    """
+    publish snapshot of snapshot
+    """
+    fixtureGpg = True
+
+    def check(self):
+        repo_name = self.random_name()
+        self.check_equal(self.post(
+            "/api/repos", json={"Name": repo_name, "DefaultDistribution": "wheezy"}).status_code, 201)
+
+        d = self.random_name()
+        self.check_equal(
+            self.upload("/api/files/" + d,
+                        "pyspi_0.6.1-1.3.dsc",
+                        "pyspi_0.6.1-1.3.diff.gz", "pyspi_0.6.1.orig.tar.gz",
+                        "pyspi-0.6.1-1.3.stripped.dsc").status_code, 200)
+        task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
+        self.check_task(task)
+
+        snapshot1_name = self.random_name()
+        task = self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot1_name})
+        self.check_task(task)
+
+        prefix = self.random_name()
+        task = self.post_task(
+            "/api/publish/" + prefix,
+            json={
+                "Architectures": ["i386", "source"],
+                "SourceKind": "snapshot",
+                "Sources": [{"Name": snapshot1_name}],
+                "Signing": DefaultSigningOptions,
+            })
+        self.check_task(task)
+
+        repo_expected = {
+            'AcquireByHash': False,
+            'Architectures': ['i386', 'source'],
+            'Codename': '',
+            'Distribution': 'wheezy',
+            'Label': '',
+            'NotAutomatic': '',
+            'ButAutomaticUpgrades': '',
+            'Origin': '',
+            'Version': '',
+            'Path': prefix + '/' + 'wheezy',
+            'Prefix': prefix,
+            'SignedBy': '',
+            'SkipContents': False,
+            'MultiDist': False,
+            'SourceKind': 'snapshot',
+            'Sources': [{'Component': 'main', 'Name': snapshot1_name}],
+            'Storage': '',
+            'Suite': ''}
+        all_repos = self.get("/api/publish")
+        self.check_equal(all_repos.status_code, 200)
+        self.check_in(repo_expected, all_repos.json())
+
+        self.check_not_exists(
+            "public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
+        self.check_exists("public/" + prefix +
+                          "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
+
+        snapshot2_name = self.random_name()
+        task = self.post_task("/api/snapshots", json={"Name": snapshot2_name, 'SourceSnapshots': [snapshot1_name]})
+        self.check_task(task)
+
+        task = self.put_task(
+            "/api/publish/" + prefix + "/wheezy",
+            json={
+                "Snapshots": [{"Component": "main", "Name": snapshot2_name}],
+                "Signing": DefaultSigningOptions,
+                "SkipContents": True,
+                "Label": "fun",
+                "Origin": "earth",
+                "Version": "13.3",
+            })
+        self.check_task(task)
+        repo_expected = {
+            'AcquireByHash': False,
+            'Architectures': ['i386', 'source'],
+            'Codename': '',
+            'Distribution': 'wheezy',
+            'Label': 'fun',
+            'Origin': 'earth',
+            'Version': '13.3',
+            'NotAutomatic': '',
+            'ButAutomaticUpgrades': '',
+            'Path': prefix + '/' + 'wheezy',
+            'Prefix': prefix,
+            'SignedBy': '',
+            'SkipContents': True,
+            'MultiDist': False,
+            'SourceKind': 'snapshot',
+            'Sources': [{'Component': 'main', 'Name': snapshot2_name}],
+            'Storage': '',
+            'Suite': ''}
+
+        all_repos = self.get("/api/publish")
+        self.check_equal(all_repos.status_code, 200)
+        self.check_in(repo_expected, all_repos.json())
+
+        # FIXME: what should exist here ? publish snapshot of snapshot
+        self.check_not_exists(
+            "public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
+        self.check_not_exists("public/" + prefix +
+                              "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
+
+        task = self.delete_task("/api/publish/" + prefix + "/wheezy")
+        self.check_task(task)
+        self.check_not_exists("public/" + prefix + "dists/")
+
+
 class PublishSwitchAPITestRepoSignedBy(APITest):
     """
     PUT /publish/:prefix/:distribution (snapshots), DELETE /publish/:prefix/:distribution

system/t12_api/unix_socket.py

@@ -14,7 +14,7 @@ class UnixSocketAPITest(BaseTest):
     socket_path = "/tmp/_aptly_test.sock"
     base_url = ("unix://%s" % socket_path)
     aptly_out = None
-    debugOutput = True
+    debugOutput = False  # Controlled by --debug flag in run.py
 
     def prepare(self):
         if self.aptly_server is None:

task/list.go

@@ -65,6 +65,7 @@ func (list *List) consumer() {
 						task.State = SUCCEEDED
 					}
 
+					fmt.Printf("RACE DEBUG: Task Done     '%s', freeing %s\n", task.Name, task.resources)
 					list.usedResources.Free(task.resources)
 
 					task.wgTask.Done()
@@ -77,6 +78,8 @@ func (list *List) consumer() {
 							blockingTasks := list.usedResources.UsedBy(t.resources)
 							if len(blockingTasks) == 0 {
 								list.usedResources.MarkInUse(t.resources, t)
+
+								fmt.Printf("RACE DEBUG: Task Resuming '%s', locking %s\n", t.Name, t.resources)
 								// unlock list since queueing may block
 								list.Unlock()
 								unlocked = true
@@ -209,10 +212,12 @@ func (list *List) RunTaskInBackground(name string, resources []string, process P
 	tasks := list.usedResources.UsedBy(resources)
 	if len(tasks) == 0 {
 		list.usedResources.MarkInUse(task.resources, task)
+		fmt.Printf("RACE DEBUG: Task Starting '%s', locking %s\n", name, resources)
 		// queueing task might block if channel not ready, unlock list before queueing
 		list.Unlock()
 		list.queue <- task
 	} else {
+		fmt.Printf("RACE DEBUG: Task Queued   '%s', waiting on %s\n", name, resources)
 		list.Unlock()
 	}