@@ -12,12 +12,47 @@ import java.util.concurrent.{ExecutorService, Executors, TimeUnit}
1212import scala .collection .mutable .ListBuffer
1313import scala .concurrent .duration .Duration
1414import scala .concurrent .{Await , ExecutionContext , Future , blocking }
15+ import scala .util .{Failure , Try }
1516import scala .util .control .NonFatal
1617
1718trait HasFabricNotebookTestConnection extends HasFabricOperationsConnection {
1819 fabricClientId = Some (FabricTestConstants .INTEGRATION_APP_ID )
1920 fabricRedirectUri = Some (FabricTestConstants .INTEGRATION_REDIRECT_URI )
20- fabricWorkspaceId = Some (FabricTestConstants .INTEGRATION_WORKSPACE_ID )
21+
22+ protected def integrationWorkspaceId : String = FabricTestConstants .INTEGRATION_WORKSPACE_ID
23+
24+ protected def cleanupStaleArtifacts (): Unit = {
25+ val dryRun = sys.env.getOrElse(" SYNAPSEML_FABRIC_CLEANUP_DRY_RUN" , " false" )
26+ require(Set (" true" , " false" )(dryRun), " SYNAPSEML_FABRIC_CLEANUP_DRY_RUN must be true or false" )
27+ fabricWorkspaceId = Some (integrationWorkspaceId)
28+ fabric.cleanupTestArtifacts(dryRun.toBoolean)
29+ }
30+
31+ protected final def captureFabricSetup [T ](setup : => T ): Try [T ] = {
32+ try {
33+ Try (setup)
34+ } catch {
35+ case error : InterruptedException => Failure (error)
36+ }
37+ }
38+
39+ protected final def getFabricSetup [T ](setup : Try [T ]): T = setup match {
40+ case Failure (error : InterruptedException ) =>
41+ Thread .currentThread().interrupt()
42+ throw error
43+ case _ => setup.get
44+ }
45+
46+ private lazy val preflight = captureFabricSetup(cleanupStaleArtifacts())
47+
48+ protected final def ensureFabricPreflight (): Unit = getFabricSetup(preflight)
49+
50+ private lazy val storeSetup = captureFabricSetup {
51+ ensureFabricPreflight()
52+ createTrackedStore()
53+ }
54+
55+ protected final def preparedStore : String = getFabricSetup(storeSetup)
2156
2257 private val artifactTracker =
2358 new FabricTestArtifactTracker (artifactId => fabric.deleteArtifact(artifactId))
@@ -28,13 +63,13 @@ trait HasFabricNotebookTestConnection extends HasFabricOperationsConnection {
2863 artifactTracker.withArtifact(artifactId)(use)
2964
3065 protected def cleanupTrackedArtifacts (): Unit = artifactTracker.cleanup()
66+
67+ protected def createTrackedStore (): String = trackArtifact(fabric.createStoreArtifact())
3168}
3269
3370class FabricTestCleanup extends TestBase with HasFabricNotebookTestConnection {
3471 test(" Clean up owned Fabric test artifacts older than 24 hours" ) {
35- val dryRun = sys.env.getOrElse(" SYNAPSEML_FABRIC_CLEANUP_DRY_RUN" , " false" )
36- require(Set (" true" , " false" )(dryRun), " SYNAPSEML_FABRIC_CLEANUP_DRY_RUN must be true or false" )
37- fabric.cleanupTestArtifacts(dryRun.toBoolean)
72+ ensureFabricPreflight()
3873 }
3974}
4075
@@ -63,13 +98,18 @@ class FabricSmokeTests extends TestBase with HasFabricNotebookTestConnection {
6398 f
6499 }
65100
66- val storeArtifactId : String = trackArtifact(fabric.createStoreArtifact())
101+ lazy val storeArtifactId : String = preparedStore
67102
68103 test(" OnePlusOne" ) {
104+ ensureFabricPreflight()
105+ runSmokeTest(storeArtifactId)
106+ }
107+
108+ protected def runSmokeTest (storeId : String ): Unit = {
69109 val notebookName = fabric.getBlobNameFromFilepath(notebookFile.getPath)
70110 withTrackedArtifact(fabric.createSJDArtifact(notebookFile.getPath)) { artifactId =>
71111 val notebookBlobPath = fabric.uploadNotebookToAzure(notebookFile)
72- fabric.updateSJDArtifact(notebookBlobPath, artifactId, storeArtifactId , includePackages = false )
112+ fabric.updateSJDArtifact(notebookBlobPath, artifactId, storeId , includePackages = false )
73113 blocking {
74114 Thread .sleep(3000 ) // scalastyle:ignore
75115 }
@@ -99,55 +139,77 @@ class FabricSmokeTests extends TestBase with HasFabricNotebookTestConnection {
99139}
100140
101141class FabricNotebookTests extends TestBase with HasFabricNotebookTestConnection {
102- SharedNotebookE2ETestUtilities .generateNotebooks()
142+ protected def discoverNotebooks (): Array [File ] = {
143+ SharedNotebookE2ETestUtilities .generateNotebooks()
144+ FileUtilities .recursiveListFiles(SharedNotebookE2ETestUtilities .NotebooksDir )
145+ .filter(_.getAbsolutePath.endsWith(" .py" ))
146+ .filter(f => FabricNotebookTests .IncludedNotebooks .exists(f.getName.startsWith))
147+ .sortBy(_.getAbsolutePath)
148+ }
103149
104- val selectedPythonFiles : Array [File ] = FileUtilities
105- .recursiveListFiles(SharedNotebookE2ETestUtilities .NotebooksDir )
106- .filter(_.getAbsolutePath.endsWith(" .py" ))
107- .filter(f => FabricNotebookTests .IncludedNotebooks .exists(f.getName.startsWith))
108- .sortBy(_.getAbsolutePath)
150+ val selectedPythonFiles : Array [File ] = discoverNotebooks()
109151
110152 selectedPythonFiles.foreach(x => println(s " Fabric notebook to be tested: $x" ))
111153 assert(selectedPythonFiles.nonEmpty, " No notebooks found to test" )
112154
113- val storeArtifactId : String = trackArtifact(fabric.createStoreArtifact())
155+ lazy val storeArtifactId : String = preparedStore
114156
115- val executorService = Executors .newFixedThreadPool(FabricNotebookTests .MaxConcurrency )
116- implicit val executionContext : ExecutionContext = ExecutionContext .fromExecutor(executorService)
157+ @ volatile private var executorStarted = false
158+ protected def createNotebookExecutor (): ExecutorService =
159+ Executors .newFixedThreadPool(FabricNotebookTests .MaxConcurrency )
117160
118- // Submit all SJDs in parallel, each Future handles create -> upload -> submit -> monitor
119- val futures : Array [(Future [String ], String )] = selectedPythonFiles.map { notebookFile =>
120- val notebookName = fabric.getBlobNameFromFilepath(notebookFile.getPath)
121- val future = Future {
122- withTrackedArtifact(fabric.createSJDArtifact(notebookFile.getPath)) { artifactId =>
123- val notebookBlobPath = fabric.uploadNotebookToAzure(notebookFile)
124- fabric.updateSJDArtifact(notebookBlobPath, artifactId, storeArtifactId)
125- blocking { Thread .sleep(3000 ) } // scalastyle:ignore
126- val jobInstanceId = fabric.submitJob(artifactId)
127- blocking { Thread .sleep(10000 ) } // scalastyle:ignore
128- Await .result(
129- fabric.monitorJob(artifactId, jobInstanceId),
130- Duration (fabric.timeoutInMillis.toLong, TimeUnit .MILLISECONDS ))
131- }
161+ lazy val executorService : ExecutorService = {
162+ val executor = createNotebookExecutor()
163+ executorStarted = true
164+ executor
165+ }
166+ implicit lazy val executionContext : ExecutionContext = ExecutionContext .fromExecutor(executorService)
167+
168+ protected def notebookTimeout : Duration =
169+ Duration (fabric.timeoutInMillis.toLong, TimeUnit .MILLISECONDS )
170+
171+ protected def runNotebook (notebookFile : File , storeId : String ): String =
172+ withTrackedArtifact(fabric.createSJDArtifact(notebookFile.getPath)) { artifactId =>
173+ val notebookBlobPath = fabric.uploadNotebookToAzure(notebookFile)
174+ fabric.updateSJDArtifact(notebookBlobPath, artifactId, storeId)
175+ blocking { Thread .sleep(3000 ) } // scalastyle:ignore
176+ val jobInstanceId = fabric.submitJob(artifactId)
177+ blocking { Thread .sleep(10000 ) } // scalastyle:ignore
178+ Await .result(fabric.monitorJob(artifactId, jobInstanceId), notebookTimeout)
179+ }
180+
181+ // Start the existing parallel workload only after the first selected test passes preflight.
182+ private lazy val submissions = captureFabricSetup {
183+ ensureFabricPreflight()
184+ val storeId = storeArtifactId
185+ selectedPythonFiles.map { notebookFile =>
186+ (Future (runNotebook(notebookFile, storeId)), notebookFile.getName)
132187 }
133- (future, notebookName)
134188 }
135189
136- futures.foreach { case (future, notebookName) =>
190+ lazy val futures : Array [(Future [String ], String )] = getFabricSetup(submissions)
191+
192+ selectedPythonFiles.zipWithIndex.foreach { case (notebookFile, index) =>
193+ val notebookName = notebookFile.getName
137194 test(notebookName) {
195+ ensureFabricPreflight()
196+ val (future, submittedNotebookName) = futures(index)
138197 try {
139- Await .result(future, Duration (fabric.timeoutInMillis.toLong, TimeUnit . MILLISECONDS ) )
198+ Await .result(future, notebookTimeout )
140199 } catch {
141- case t : Throwable =>
142- throw new RuntimeException (s " Job failed for $notebookName" , t)
200+ case error : InterruptedException =>
201+ Thread .currentThread().interrupt()
202+ throw error
203+ case NonFatal (t) =>
204+ throw new RuntimeException (s " Job failed for $submittedNotebookName" , t)
143205 }
144206 }
145207 }
146208
147209 override def afterAll (): Unit = {
148210 try {
149211 FabricNotebookTests .shutdownAndCleanup(
150- FabricNotebookTests .shutdownExecutor(executorService),
212+ if (executorStarted) FabricNotebookTests .shutdownExecutor(executorService),
151213 cleanupTrackedArtifacts())
152214 } finally {
153215 super .afterAll()
0 commit comments