http://git-wip-us.apache.org/repos/asf/zeppelin/blob/8d4902e7/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterTest.java ---------------------------------------------------------------------- diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterTest.java deleted file mode 100644 index 95235e5..0000000 --- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterTest.java +++ /dev/null @@ -1,975 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.zeppelin.interpreter.remote; - -import static org.junit.Assert.*; - -import java.io.File; -import java.io.IOException; -import java.util.HashMap; -import java.util.LinkedList; -import java.util.List; -import java.util.Map; -import java.util.Properties; - -import org.apache.thrift.transport.TTransportException; -import org.apache.zeppelin.display.AngularObject; -import org.apache.zeppelin.display.AngularObjectRegistry; -import org.apache.zeppelin.interpreter.remote.mock.MockInterpreterEnv; -import org.apache.zeppelin.interpreter.thrift.RemoteInterpreterResultMessage; -import org.apache.zeppelin.interpreter.thrift.RemoteInterpreterService; -import org.apache.zeppelin.interpreter.thrift.RemoteInterpreterService.Client; -import org.apache.zeppelin.user.AuthenticationInfo; -import org.apache.zeppelin.display.GUI; -import org.apache.zeppelin.interpreter.*; -import org.apache.zeppelin.interpreter.InterpreterResult.Code; -import org.apache.zeppelin.interpreter.remote.mock.MockInterpreterA; -import org.apache.zeppelin.interpreter.remote.mock.MockInterpreterB; -import org.apache.zeppelin.resource.LocalResourcePool; -import org.apache.zeppelin.scheduler.Job; -import org.apache.zeppelin.scheduler.Job.Status; -import org.apache.zeppelin.scheduler.Scheduler; -import org.junit.After; -import org.junit.Before; -import org.junit.Test; -import org.mockito.Mockito; - -import com.google.gson.Gson; -import com.google.gson.reflect.TypeToken; - -public class RemoteInterpreterTest { - - - private static final String INTERPRETER_SCRIPT = - System.getProperty("os.name").startsWith("Windows") ? - "../bin/interpreter.cmd" : - "../bin/interpreter.sh"; - - private InterpreterGroup intpGroup; - private HashMap<String, String> env; - - @Before - public void setUp() throws Exception { - intpGroup = new InterpreterGroup(); - env = new HashMap<>(); - env.put("ZEPPELIN_CLASSPATH", new File("./target/test-classes").getAbsolutePath()); - } - - @After - public void tearDown() throws Exception { - intpGroup.close(); - } - - private RemoteInterpreter createMockInterpreterA(Properties p) { - return createMockInterpreterA(p, "note"); - } - - private RemoteInterpreter createMockInterpreterA(Properties p, String noteId) { - return new RemoteInterpreter( - p, - noteId, - MockInterpreterA.class.getName(), - new File(INTERPRETER_SCRIPT).getAbsolutePath(), - "fake", - "fakeRepo", - env, - 10 * 1000, - null, - null, - "anonymous", - false); - } - - private RemoteInterpreter createMockInterpreterB(Properties p) { - return createMockInterpreterB(p, "note"); - } - - private RemoteInterpreter createMockInterpreterB(Properties p, String noteId) { - return new RemoteInterpreter( - p, - noteId, - MockInterpreterB.class.getName(), - new File(INTERPRETER_SCRIPT).getAbsolutePath(), - "fake", - "fakeRepo", - env, - 10 * 1000, - null, - null, - "anonymous", - false); - } - - @Test - public void testRemoteInterperterCall() throws TTransportException, IOException { - Properties p = new Properties(); - intpGroup.put("note", new LinkedList<Interpreter>()); - - RemoteInterpreter intpA = createMockInterpreterA(p); - - intpGroup.get("note").add(intpA); - - intpA.setInterpreterGroup(intpGroup); - - RemoteInterpreter intpB = createMockInterpreterB(p); - - intpGroup.get("note").add(intpB); - intpB.setInterpreterGroup(intpGroup); - - - RemoteInterpreterProcess process = intpA.getInterpreterProcess(); - process.equals(intpB.getInterpreterProcess()); - - assertFalse(process.isRunning()); - assertEquals(0, process.getNumIdleClient()); - assertEquals(0, process.referenceCount()); - - intpA.open(); // initializa all interpreters in the same group - assertTrue(process.isRunning()); - assertEquals(1, process.getNumIdleClient()); - assertEquals(1, process.referenceCount()); - - intpA.interpret("1", - new InterpreterContext( - "note", - "id", - null, - "title", - "text", - new AuthenticationInfo(), - new HashMap<String, Object>(), - new GUI(), - new AngularObjectRegistry(intpGroup.getId(), null), - new LocalResourcePool("pool1"), - new LinkedList<InterpreterContextRunner>(), null)); - - intpB.open(); - assertEquals(1, process.referenceCount()); - - intpA.close(); - assertEquals(0, process.referenceCount()); - intpB.close(); - assertEquals(0, process.referenceCount()); - - assertFalse(process.isRunning()); - - } - - @Test - public void testExecuteIncorrectPrecode() throws TTransportException, IOException { - Properties p = new Properties(); - p.put("zeppelin.MockInterpreterA.precode", "fail test"); - intpGroup.put("note", new LinkedList<Interpreter>()); - - RemoteInterpreter intpA = createMockInterpreterA(p); - - intpGroup.get("note").add(intpA); - - intpA.setInterpreterGroup(intpGroup); - - RemoteInterpreterProcess process = intpA.getInterpreterProcess(); - - intpA.open(); - - InterpreterResult result = intpA.interpret("1", - new InterpreterContext( - "note", - "id", - null, - "title", - "text", - new AuthenticationInfo(), - new HashMap<String, Object>(), - new GUI(), - new AngularObjectRegistry(intpGroup.getId(), null), - new LocalResourcePool("pool1"), - new LinkedList<InterpreterContextRunner>(), null)); - - - - intpA.close(); - assertEquals(Code.ERROR, result.code()); - } - - @Test - public void testExecuteCorrectPrecode() throws TTransportException, IOException { - Properties p = new Properties(); - p.put("zeppelin.MockInterpreterA.precode", "2"); - intpGroup.put("note", new LinkedList<Interpreter>()); - - RemoteInterpreter intpA = createMockInterpreterA(p); - - intpGroup.get("note").add(intpA); - - intpA.setInterpreterGroup(intpGroup); - - RemoteInterpreterProcess process = intpA.getInterpreterProcess(); - - intpA.open(); - - InterpreterResult result = intpA.interpret("1", - new InterpreterContext( - "note", - "id", - null, - "title", - "text", - new AuthenticationInfo(), - new HashMap<String, Object>(), - new GUI(), - new AngularObjectRegistry(intpGroup.getId(), null), - new LocalResourcePool("pool1"), - new LinkedList<InterpreterContextRunner>(), null)); - - - - intpA.close(); - assertEquals(Code.SUCCESS, result.code()); - assertEquals("1", result.message().get(0).getData()); - } - - @Test - public void testRemoteInterperterErrorStatus() throws TTransportException, IOException { - Properties p = new Properties(); - - RemoteInterpreter intpA = createMockInterpreterA(p); - - intpGroup.put("note", new LinkedList<Interpreter>()); - intpGroup.get("note").add(intpA); - intpA.setInterpreterGroup(intpGroup); - - intpA.open(); - InterpreterResult ret = intpA.interpret("non numeric value", - new InterpreterContext( - "noteId", - "id", - null, - "title", - "text", - new AuthenticationInfo(), - new HashMap<String, Object>(), - new GUI(), - new AngularObjectRegistry(intpGroup.getId(), null), - new LocalResourcePool("pool1"), - new LinkedList<InterpreterContextRunner>(), null)); - - assertEquals(Code.ERROR, ret.code()); - } - - @Test - public void testRemoteSchedulerSharing() throws TTransportException, IOException { - Properties p = new Properties(); - intpGroup.put("note", new LinkedList<Interpreter>()); - - RemoteInterpreter intpA = new RemoteInterpreter( - p, - "note", - MockInterpreterA.class.getName(), - new File(INTERPRETER_SCRIPT).getAbsolutePath(), - "fake", - "fakeRepo", - env, - 10 * 1000, - null, - null, - "anonymous", - false); - - intpGroup.get("note").add(intpA); - intpA.setInterpreterGroup(intpGroup); - - RemoteInterpreter intpB = new RemoteInterpreter( - p, - "note", - MockInterpreterB.class.getName(), - new File(INTERPRETER_SCRIPT).getAbsolutePath(), - "fake", - "fakeRepo", - env, - 10 * 1000, - null, - null, - "anonymous", - false); - - intpGroup.get("note").add(intpB); - intpB.setInterpreterGroup(intpGroup); - - intpA.open(); - intpB.open(); - - long start = System.currentTimeMillis(); - InterpreterResult ret = intpA.interpret("500", - new InterpreterContext( - "note", - "id", - null, - "title", - "text", - new AuthenticationInfo(), - new HashMap<String, Object>(), - new GUI(), - new AngularObjectRegistry(intpGroup.getId(), null), - new LocalResourcePool("pool1"), - new LinkedList<InterpreterContextRunner>(), null)); - assertEquals("500", ret.message().get(0).getData()); - - ret = intpB.interpret("500", - new InterpreterContext( - "note", - "id", - null, - "title", - "text", - new AuthenticationInfo(), - new HashMap<String, Object>(), - new GUI(), - new AngularObjectRegistry(intpGroup.getId(), null), - new LocalResourcePool("pool1"), - new LinkedList<InterpreterContextRunner>(), null)); - assertEquals("1000", ret.message().get(0).getData()); - long end = System.currentTimeMillis(); - assertTrue(end - start >= 1000); - - - intpA.close(); - intpB.close(); - } - - @Test - public void testRemoteSchedulerSharingSubmit() throws TTransportException, IOException, InterruptedException { - Properties p = new Properties(); - intpGroup.put("note", new LinkedList<Interpreter>()); - - final RemoteInterpreter intpA = createMockInterpreterA(p); - - intpGroup.get("note").add(intpA); - intpA.setInterpreterGroup(intpGroup); - - final RemoteInterpreter intpB = createMockInterpreterB(p); - - intpGroup.get("note").add(intpB); - intpB.setInterpreterGroup(intpGroup); - - intpA.open(); - intpB.open(); - - long start = System.currentTimeMillis(); - Job jobA = new Job("jobA", null) { - private Object r; - - @Override - public Object getReturn() { - return r; - } - - @Override - public void setResult(Object results) { - this.r = results; - } - - @Override - public int progress() { - return 0; - } - - @Override - public Map<String, Object> info() { - return null; - } - - @Override - protected Object jobRun() throws Throwable { - return intpA.interpret("500", - new InterpreterContext( - "note", - "jobA", - null, - "title", - "text", - new AuthenticationInfo(), - new HashMap<String, Object>(), - new GUI(), - new AngularObjectRegistry(intpGroup.getId(), null), - new LocalResourcePool("pool1"), - new LinkedList<InterpreterContextRunner>(), null)); - } - - @Override - protected boolean jobAbort() { - return false; - } - - }; - intpA.getScheduler().submit(jobA); - - Job jobB = new Job("jobB", null) { - - private Object r; - - @Override - public Object getReturn() { - return r; - } - - @Override - public void setResult(Object results) { - this.r = results; - } - - @Override - public int progress() { - return 0; - } - - @Override - public Map<String, Object> info() { - return null; - } - - @Override - protected Object jobRun() throws Throwable { - return intpB.interpret("500", - new InterpreterContext( - "note", - "jobB", - null, - "title", - "text", - new AuthenticationInfo(), - new HashMap<String, Object>(), - new GUI(), - new AngularObjectRegistry(intpGroup.getId(), null), - new LocalResourcePool("pool1"), - new LinkedList<InterpreterContextRunner>(), null)); - } - - @Override - protected boolean jobAbort() { - return false; - } - - }; - intpB.getScheduler().submit(jobB); - // wait until both job finished - while (jobA.getStatus() != Status.FINISHED || - jobB.getStatus() != Status.FINISHED) { - Thread.sleep(100); - } - long end = System.currentTimeMillis(); - assertTrue(end - start >= 1000); - - assertEquals("1000", ((InterpreterResult) jobB.getReturn()).message().get(0).getData()); - - intpA.close(); - intpB.close(); - } - - @Test - public void testRunOrderPreserved() throws InterruptedException { - Properties p = new Properties(); - intpGroup.put("note", new LinkedList<Interpreter>()); - - final RemoteInterpreter intpA = createMockInterpreterA(p); - - intpGroup.get("note").add(intpA); - intpA.setInterpreterGroup(intpGroup); - - intpA.open(); - - int concurrency = 3; - final List<InterpreterResultMessage> results = new LinkedList<>(); - - Scheduler scheduler = intpA.getScheduler(); - for (int i = 0; i < concurrency; i++) { - final String jobId = Integer.toString(i); - scheduler.submit(new Job(jobId, Integer.toString(i), null, 200) { - private Object r; - - @Override - public Object getReturn() { - return r; - } - - @Override - public void setResult(Object results) { - this.r = results; - } - - @Override - public int progress() { - return 0; - } - - @Override - public Map<String, Object> info() { - return null; - } - - @Override - protected Object jobRun() throws Throwable { - InterpreterResult ret = intpA.interpret(getJobName(), new InterpreterContext( - "note", - jobId, - null, - "title", - "text", - new AuthenticationInfo(), - new HashMap<String, Object>(), - new GUI(), - new AngularObjectRegistry(intpGroup.getId(), null), - new LocalResourcePool("pool1"), - new LinkedList<InterpreterContextRunner>(), null)); - - synchronized (results) { - results.addAll(ret.message()); - results.notify(); - } - return null; - } - - @Override - protected boolean jobAbort() { - return false; - } - - }); - } - - // wait for job finished - synchronized (results) { - while (results.size() != concurrency) { - results.wait(300); - } - } - - int i = 0; - for (InterpreterResultMessage result : results) { - assertEquals(Integer.toString(i++), result.getData()); - } - assertEquals(concurrency, i); - - intpA.close(); - } - - - @Test - public void testRunParallel() throws InterruptedException { - Properties p = new Properties(); - p.put("parallel", "true"); - intpGroup.put("note", new LinkedList<Interpreter>()); - - final RemoteInterpreter intpA = createMockInterpreterA(p); - - intpGroup.get("note").add(intpA); - intpA.setInterpreterGroup(intpGroup); - - intpA.open(); - - int concurrency = 4; - final int timeToSleep = 1000; - final List<InterpreterResultMessage> results = new LinkedList<>(); - long start = System.currentTimeMillis(); - - Scheduler scheduler = intpA.getScheduler(); - for (int i = 0; i < concurrency; i++) { - final String jobId = Integer.toString(i); - scheduler.submit(new Job(jobId, Integer.toString(i), null, 300) { - private Object r; - - @Override - public Object getReturn() { - return r; - } - - @Override - public void setResult(Object results) { - this.r = results; - } - - @Override - public int progress() { - return 0; - } - - @Override - public Map<String, Object> info() { - return null; - } - - @Override - protected Object jobRun() throws Throwable { - String stmt = Integer.toString(timeToSleep); - InterpreterResult ret = intpA.interpret(stmt, new InterpreterContext( - "note", - jobId, - null, - "title", - "text", - new AuthenticationInfo(), - new HashMap<String, Object>(), - new GUI(), - new AngularObjectRegistry(intpGroup.getId(), null), - new LocalResourcePool("pool1"), - new LinkedList<InterpreterContextRunner>(), null)); - - synchronized (results) { - results.addAll(ret.message()); - results.notify(); - } - return stmt; - } - - @Override - protected boolean jobAbort() { - return false; - } - - }); - } - - // wait for job finished - synchronized (results) { - while (results.size() != concurrency) { - results.wait(300); - } - } - - long end = System.currentTimeMillis(); - - assertTrue(end - start < timeToSleep * concurrency); - - intpA.close(); - } - - @Test - public void testInterpreterGroupResetBeforeProcessStarts() { - Properties p = new Properties(); - - RemoteInterpreter intpA = createMockInterpreterA(p); - - intpA.setInterpreterGroup(intpGroup); - RemoteInterpreterProcess processA = intpA.getInterpreterProcess(); - - intpA.setInterpreterGroup(new InterpreterGroup(intpA.getInterpreterGroup().getId())); - RemoteInterpreterProcess processB = intpA.getInterpreterProcess(); - - assertNotSame(processA.hashCode(), processB.hashCode()); - } - - @Test - public void testInterpreterGroupResetAfterProcessFinished() { - Properties p = new Properties(); - intpGroup.put("note", new LinkedList<Interpreter>()); - - RemoteInterpreter intpA = createMockInterpreterA(p); - - intpA.setInterpreterGroup(intpGroup); - RemoteInterpreterProcess processA = intpA.getInterpreterProcess(); - intpA.open(); - - processA.dereference(); // intpA.close(); - - intpA.setInterpreterGroup(new InterpreterGroup(intpA.getInterpreterGroup().getId())); - RemoteInterpreterProcess processB = intpA.getInterpreterProcess(); - - assertNotSame(processA.hashCode(), processB.hashCode()); - } - - @Test - public void testInterpreterGroupResetDuringProcessRunning() throws InterruptedException { - Properties p = new Properties(); - intpGroup.put("note", new LinkedList<Interpreter>()); - - final RemoteInterpreter intpA = createMockInterpreterA(p); - - intpGroup.get("note").add(intpA); - intpA.setInterpreterGroup(intpGroup); - - intpA.open(); - - Job jobA = new Job("jobA", null) { - private Object r; - - @Override - public Object getReturn() { - return r; - } - - @Override - public void setResult(Object results) { - this.r = results; - } - - @Override - public int progress() { - return 0; - } - - @Override - public Map<String, Object> info() { - return null; - } - - @Override - protected Object jobRun() throws Throwable { - return intpA.interpret("2000", - new InterpreterContext( - "note", - "jobA", - null, - "title", - "text", - new AuthenticationInfo(), - new HashMap<String, Object>(), - new GUI(), - new AngularObjectRegistry(intpGroup.getId(), null), - new LocalResourcePool("pool1"), - new LinkedList<InterpreterContextRunner>(), null)); - } - - @Override - protected boolean jobAbort() { - return false; - } - - }; - intpA.getScheduler().submit(jobA); - - // wait for job started - while (intpA.getScheduler().getJobsRunning().size() == 0) { - Thread.sleep(100); - } - - // restart interpreter - RemoteInterpreterProcess processA = intpA.getInterpreterProcess(); - intpA.close(); - - InterpreterGroup newInterpreterGroup = - new InterpreterGroup(intpA.getInterpreterGroup().getId()); - newInterpreterGroup.put("note", new LinkedList<Interpreter>()); - - intpA.setInterpreterGroup(newInterpreterGroup); - intpA.open(); - RemoteInterpreterProcess processB = intpA.getInterpreterProcess(); - - assertNotSame(processA.hashCode(), processB.hashCode()); - - } - - @Test - public void testRemoteInterpreterSharesTheSameSchedulerInstanceInTheSameGroup() { - Properties p = new Properties(); - intpGroup.put("note", new LinkedList<Interpreter>()); - - RemoteInterpreter intpA = createMockInterpreterA(p); - - intpGroup.get("note").add(intpA); - intpA.setInterpreterGroup(intpGroup); - - RemoteInterpreter intpB = createMockInterpreterB(p); - - intpGroup.get("note").add(intpB); - intpB.setInterpreterGroup(intpGroup); - - intpA.open(); - intpB.open(); - - assertEquals(intpA.getScheduler(), intpB.getScheduler()); - } - - @Test - public void testMultiInterpreterSession() { - Properties p = new Properties(); - intpGroup.put("sessionA", new LinkedList<Interpreter>()); - intpGroup.put("sessionB", new LinkedList<Interpreter>()); - - RemoteInterpreter intpAsessionA = createMockInterpreterA(p, "sessionA"); - intpGroup.get("sessionA").add(intpAsessionA); - intpAsessionA.setInterpreterGroup(intpGroup); - - RemoteInterpreter intpBsessionA = createMockInterpreterB(p, "sessionA"); - intpGroup.get("sessionA").add(intpBsessionA); - intpBsessionA.setInterpreterGroup(intpGroup); - - intpAsessionA.open(); - intpBsessionA.open(); - - assertEquals(intpAsessionA.getScheduler(), intpBsessionA.getScheduler()); - - RemoteInterpreter intpAsessionB = createMockInterpreterA(p, "sessionB"); - intpGroup.get("sessionB").add(intpAsessionB); - intpAsessionB.setInterpreterGroup(intpGroup); - - RemoteInterpreter intpBsessionB = createMockInterpreterB(p, "sessionB"); - intpGroup.get("sessionB").add(intpBsessionB); - intpBsessionB.setInterpreterGroup(intpGroup); - - intpAsessionB.open(); - intpBsessionB.open(); - - assertEquals(intpAsessionB.getScheduler(), intpBsessionB.getScheduler()); - assertNotEquals(intpAsessionA.getScheduler(), intpAsessionB.getScheduler()); - } - - @Test - public void should_push_local_angular_repo_to_remote() throws Exception { - //Given - final Client client = Mockito.mock(Client.class); - final RemoteInterpreter intr = new RemoteInterpreter(new Properties(), "noteId", - MockInterpreterA.class.getName(), "runner", "path", "localRepo", env, 10 * 1000, null, - null, "anonymous", false); - final AngularObjectRegistry registry = new AngularObjectRegistry("spark", null); - registry.add("name", "DuyHai DOAN", "nodeId", "paragraphId"); - final InterpreterGroup interpreterGroup = new InterpreterGroup("groupId"); - interpreterGroup.setAngularObjectRegistry(registry); - intr.setInterpreterGroup(interpreterGroup); - - final java.lang.reflect.Type registryType = new TypeToken<Map<String, - Map<String, AngularObject>>>() {}.getType(); - final Gson gson = new Gson(); - final String expected = gson.toJson(registry.getRegistry(), registryType); - - //When - intr.pushAngularObjectRegistryToRemote(client); - - //Then - Mockito.verify(client).angularRegistryPush(expected); - } - - @Test - public void testEnvStringPattern() { - assertFalse(RemoteInterpreterUtils.isEnvString(null)); - assertFalse(RemoteInterpreterUtils.isEnvString("")); - assertFalse(RemoteInterpreterUtils.isEnvString("abcDEF")); - assertFalse(RemoteInterpreterUtils.isEnvString("ABC-DEF")); - assertTrue(RemoteInterpreterUtils.isEnvString("ABCDEF")); - assertTrue(RemoteInterpreterUtils.isEnvString("ABC_DEF")); - assertTrue(RemoteInterpreterUtils.isEnvString("ABC_DEF123")); - } - - @Test - public void testEnvronmentAndPropertySet() { - Properties p = new Properties(); - p.setProperty("MY_ENV1", "env value 1"); - p.setProperty("my.property.1", "property value 1"); - - RemoteInterpreter intp = new RemoteInterpreter( - p, - "note", - MockInterpreterEnv.class.getName(), - new File(INTERPRETER_SCRIPT).getAbsolutePath(), - "fake", - "fakeRepo", - env, - 10 * 1000, - null, - null, - "anonymous", - false); - - intpGroup.put("note", new LinkedList<Interpreter>()); - intpGroup.get("note").add(intp); - intp.setInterpreterGroup(intpGroup); - - intp.open(); - - InterpreterContext context = new InterpreterContext( - "noteId", - "id", - null, - "title", - "text", - new AuthenticationInfo(), - new HashMap<String, Object>(), - new GUI(), - new AngularObjectRegistry(intpGroup.getId(), null), - new LocalResourcePool("pool1"), - new LinkedList<InterpreterContextRunner>(), null); - - - assertEquals("env value 1", intp.interpret("getEnv MY_ENV1", context).message().get(0).getData()); - assertEquals(Code.ERROR, intp.interpret("getProperty MY_ENV1", context).code()); - assertEquals(Code.ERROR, intp.interpret("getEnv my.property.1", context).code()); - assertEquals("property value 1", intp.interpret("getProperty my.property.1", context).message().get(0).getData()); - - intp.close(); - } - - @Test - public void testSetProgress() throws InterruptedException { - // given MockInterpreterA set progress through InterpreterContext - Properties p = new Properties(); - p.setProperty("progress", "50"); - final RemoteInterpreter intpA = createMockInterpreterA(p); - - intpGroup.put("note", new LinkedList<Interpreter>()); - intpGroup.get("note").add(intpA); - intpA.setInterpreterGroup(intpGroup); - - intpA.open(); - - final InterpreterContext context1 = new InterpreterContext( - "noteId", - "id1", - null, - "title", - "text", - new AuthenticationInfo(), - new HashMap<String, Object>(), - new GUI(), - new AngularObjectRegistry(intpGroup.getId(), null), - new LocalResourcePool("pool1"), - new LinkedList<InterpreterContextRunner>(), null); - - InterpreterContext context2 = new InterpreterContext( - "noteId", - "id2", - null, - "title", - "text", - new AuthenticationInfo(), - new HashMap<String, Object>(), - new GUI(), - new AngularObjectRegistry(intpGroup.getId(), null), - new LocalResourcePool("pool1"), - new LinkedList<InterpreterContextRunner>(), null); - - - assertEquals(0, intpA.getProgress(context1)); - assertEquals(0, intpA.getProgress(context2)); - - // when interpreter update progress through InterpreterContext - Thread t = new Thread() { - public void run() { - InterpreterResult ret = intpA.interpret("1000", context1); - } - }; - t.start(); - - // then progress need to be updated in given context - while(intpA.getProgress(context1) == 0) Thread.yield(); - assertEquals(50, intpA.getProgress(context1)); - assertEquals(0, intpA.getProgress(context2)); - - t.join(); - assertEquals(0, intpA.getProgress(context1)); - assertEquals(0, intpA.getProgress(context2)); - } - -}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/8d4902e7/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterUtilsTest.java ---------------------------------------------------------------------- diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterUtilsTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterUtilsTest.java deleted file mode 100644 index 975d6ea..0000000 --- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterUtilsTest.java +++ /dev/null @@ -1,34 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.zeppelin.interpreter.remote; - -import static org.junit.Assert.assertTrue; - -import java.io.IOException; - -import org.apache.zeppelin.interpreter.remote.RemoteInterpreterUtils; -import org.junit.Test; - -public class RemoteInterpreterUtilsTest { - - @Test - public void testFindRandomAvailablePortOnAllLocalInterfaces() throws IOException { - assertTrue(RemoteInterpreterUtils.findRandomAvailablePortOnAllLocalInterfaces() > 0); - } - -} http://git-wip-us.apache.org/repos/asf/zeppelin/blob/8d4902e7/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterA.java ---------------------------------------------------------------------- diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterA.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterA.java deleted file mode 100644 index 50d9888..0000000 --- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterA.java +++ /dev/null @@ -1,97 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.zeppelin.interpreter.remote.mock; - -import java.util.List; -import java.util.Properties; - -import org.apache.zeppelin.interpreter.Interpreter; -import org.apache.zeppelin.interpreter.InterpreterContext; -import org.apache.zeppelin.interpreter.InterpreterException; -import org.apache.zeppelin.interpreter.InterpreterPropertyBuilder; -import org.apache.zeppelin.interpreter.InterpreterResult; -import org.apache.zeppelin.interpreter.InterpreterResult.Code; -import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion; -import org.apache.zeppelin.scheduler.Scheduler; -import org.apache.zeppelin.scheduler.SchedulerFactory; - -public class MockInterpreterA extends Interpreter { - - private String lastSt; - - public MockInterpreterA(Properties property) { - super(property); - } - - @Override - public void open() { - //new RuntimeException().printStackTrace(); - } - - @Override - public void close() { - } - - public String getLastStatement() { - return lastSt; - } - - @Override - public InterpreterResult interpret(String st, InterpreterContext context) { - if (property.containsKey("progress")) { - context.setProgress(Integer.parseInt(getProperty("progress"))); - } - try { - Thread.sleep(Long.parseLong(st)); - this.lastSt = st; - } catch (NumberFormatException | InterruptedException e) { - throw new InterpreterException(e); - } - return new InterpreterResult(Code.SUCCESS, st); - } - - @Override - public void cancel(InterpreterContext context) { - - } - - @Override - public FormType getFormType() { - return FormType.NATIVE; - } - - @Override - public int getProgress(InterpreterContext context) { - return 0; - } - - @Override - public List<InterpreterCompletion> completion(String buf, int cursor, - InterpreterContext interpreterContext) { - return null; - } - - @Override - public Scheduler getScheduler() { - if (getProperty("parallel") != null && getProperty("parallel").equals("true")) { - return SchedulerFactory.singleton().createOrGetParallelScheduler("interpreter_" + this.hashCode(), 10); - } else { - return SchedulerFactory.singleton().createOrGetFIFOScheduler("interpreter_" + this.hashCode()); - } - } -} http://git-wip-us.apache.org/repos/asf/zeppelin/blob/8d4902e7/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterAngular.java ---------------------------------------------------------------------- diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterAngular.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterAngular.java deleted file mode 100644 index d4b26ad..0000000 --- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterAngular.java +++ /dev/null @@ -1,113 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.zeppelin.interpreter.remote.mock; - -import java.util.List; -import java.util.Properties; -import java.util.concurrent.atomic.AtomicInteger; - -import org.apache.zeppelin.display.AngularObjectRegistry; -import org.apache.zeppelin.display.AngularObjectWatcher; -import org.apache.zeppelin.interpreter.Interpreter; -import org.apache.zeppelin.interpreter.InterpreterContext; -import org.apache.zeppelin.interpreter.InterpreterPropertyBuilder; -import org.apache.zeppelin.interpreter.InterpreterResult; -import org.apache.zeppelin.interpreter.InterpreterResult.Code; -import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion; - -public class MockInterpreterAngular extends Interpreter { - - AtomicInteger numWatch = new AtomicInteger(0); - - public MockInterpreterAngular(Properties property) { - super(property); - } - - @Override - public void open() { - } - - @Override - public void close() { - - } - - @Override - public InterpreterResult interpret(String st, InterpreterContext context) { - String[] stmt = st.split(" "); - String cmd = stmt[0]; - String name = null; - if (stmt.length >= 2) { - name = stmt[1]; - } - String value = null; - if (stmt.length == 3) { - value = stmt[2]; - } - - AngularObjectRegistry registry = context.getAngularObjectRegistry(); - - if (cmd.equals("add")) { - registry.add(name, value, context.getNoteId(), null); - registry.get(name, context.getNoteId(), null).addWatcher(new AngularObjectWatcher - (null) { - - @Override - public void watch(Object oldObject, Object newObject, - InterpreterContext context) { - numWatch.incrementAndGet(); - } - - }); - } else if (cmd.equalsIgnoreCase("update")) { - registry.get(name, context.getNoteId(), null).set(value); - } else if (cmd.equals("remove")) { - registry.remove(name, context.getNoteId(), null); - } - - try { - Thread.sleep(500); // wait for watcher executed - } catch (InterruptedException e) { - logger.error("Exception in MockInterpreterAngular while interpret Thread.sleep", e); - } - - String msg = registry.getAll(context.getNoteId(), null).size() + " " + Integer.toString(numWatch - .get()); - return new InterpreterResult(Code.SUCCESS, msg); - } - - @Override - public void cancel(InterpreterContext context) { - } - - @Override - public FormType getFormType() { - return FormType.NATIVE; - } - - @Override - public int getProgress(InterpreterContext context) { - return 0; - } - - @Override - public List<InterpreterCompletion> completion(String buf, int cursor, - InterpreterContext interpreterContext) { - return null; - } -} http://git-wip-us.apache.org/repos/asf/zeppelin/blob/8d4902e7/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterB.java ---------------------------------------------------------------------- diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterB.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterB.java deleted file mode 100644 index 7103335..0000000 --- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterB.java +++ /dev/null @@ -1,126 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.zeppelin.interpreter.remote.mock; - -import java.util.List; -import java.util.Properties; - -import org.apache.zeppelin.interpreter.Interpreter; -import org.apache.zeppelin.interpreter.InterpreterContext; -import org.apache.zeppelin.interpreter.InterpreterException; -import org.apache.zeppelin.interpreter.InterpreterGroup; -import org.apache.zeppelin.interpreter.InterpreterPropertyBuilder; -import org.apache.zeppelin.interpreter.InterpreterResult; -import org.apache.zeppelin.interpreter.InterpreterResult.Code; -import org.apache.zeppelin.interpreter.WrappedInterpreter; -import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion; -import org.apache.zeppelin.scheduler.Scheduler; - -public class MockInterpreterB extends Interpreter { - - public MockInterpreterB(Properties property) { - super(property); - } - - @Override - public void open() { - //new RuntimeException().printStackTrace(); - } - - @Override - public void close() { - } - - @Override - public InterpreterResult interpret(String st, InterpreterContext context) { - MockInterpreterA intpA = getInterpreterA(); - String intpASt = intpA.getLastStatement(); - long timeToSleep = Long.parseLong(st); - if (intpASt != null) { - timeToSleep += Long.parseLong(intpASt); - } - try { - Thread.sleep(timeToSleep); - } catch (NumberFormatException | InterruptedException e) { - throw new InterpreterException(e); - } - return new InterpreterResult(Code.SUCCESS, Long.toString(timeToSleep)); - } - - @Override - public void cancel(InterpreterContext context) { - - } - - @Override - public FormType getFormType() { - return FormType.NATIVE; - } - - @Override - public int getProgress(InterpreterContext context) { - return 0; - } - - @Override - public List<InterpreterCompletion> completion(String buf, int cursor, - InterpreterContext interpreterContext) { - return null; - } - - public MockInterpreterA getInterpreterA() { - InterpreterGroup interpreterGroup = getInterpreterGroup(); - synchronized (interpreterGroup) { - for (List<Interpreter> interpreters : interpreterGroup.values()) { - boolean belongsToSameNoteGroup = false; - MockInterpreterA a = null; - for (Interpreter intp : interpreters) { - if (intp.getClassName().equals(MockInterpreterA.class.getName())) { - Interpreter p = intp; - while (p instanceof WrappedInterpreter) { - p = ((WrappedInterpreter) p).getInnerInterpreter(); - } - a = (MockInterpreterA) p; - } - - Interpreter p = intp; - while (p instanceof WrappedInterpreter) { - p = ((WrappedInterpreter) p).getInnerInterpreter(); - } - if (this == p) { - belongsToSameNoteGroup = true; - } - } - if (belongsToSameNoteGroup) { - return a; - } - } - } - return null; - } - - @Override - public Scheduler getScheduler() { - MockInterpreterA intpA = getInterpreterA(); - if (intpA != null) { - return intpA.getScheduler(); - } - return null; - } - -} http://git-wip-us.apache.org/repos/asf/zeppelin/blob/8d4902e7/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterEnv.java ---------------------------------------------------------------------- diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterEnv.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterEnv.java deleted file mode 100644 index 12e11f7..0000000 --- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterEnv.java +++ /dev/null @@ -1,80 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.zeppelin.interpreter.remote.mock; - -import org.apache.zeppelin.interpreter.*; -import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion; -import org.apache.zeppelin.scheduler.Scheduler; -import org.apache.zeppelin.scheduler.SchedulerFactory; - -import java.util.List; -import java.util.Properties; - - -public class MockInterpreterEnv extends Interpreter { - - public MockInterpreterEnv(Properties property) { - super(property); - } - - @Override - public void open() { - } - - @Override - public void close() { - } - - @Override - public InterpreterResult interpret(String st, InterpreterContext context) { - String[] cmd = st.split(" "); - if (cmd[0].equals("getEnv")) { - return new InterpreterResult(InterpreterResult.Code.SUCCESS, System.getenv(cmd[1])); - } else if (cmd[0].equals("getProperty")){ - return new InterpreterResult(InterpreterResult.Code.SUCCESS, System.getProperty(cmd[1])); - } else { - return new InterpreterResult(InterpreterResult.Code.ERROR, cmd[0]); - } - } - - @Override - public void cancel(InterpreterContext context) { - - } - - @Override - public FormType getFormType() { - return FormType.NATIVE; - } - - @Override - public int getProgress(InterpreterContext context) { - return 0; - } - - @Override - public List<InterpreterCompletion> completion(String buf, int cursor, - InterpreterContext interpreterContext) { - return null; - } - - @Override - public Scheduler getScheduler() { - return SchedulerFactory.singleton().createOrGetFIFOScheduler("interpreter_" + this.hashCode()); - } -} - http://git-wip-us.apache.org/repos/asf/zeppelin/blob/8d4902e7/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterOutputStream.java ---------------------------------------------------------------------- diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterOutputStream.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterOutputStream.java deleted file mode 100644 index 349315c..0000000 --- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterOutputStream.java +++ /dev/null @@ -1,90 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.zeppelin.interpreter.remote.mock; - -import org.apache.zeppelin.interpreter.*; -import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion; -import org.apache.zeppelin.scheduler.Scheduler; -import org.apache.zeppelin.scheduler.SchedulerFactory; - -import java.io.IOException; -import java.util.List; -import java.util.Properties; - -/** - * MockInterpreter to test outputstream - */ -public class MockInterpreterOutputStream extends Interpreter { - private String lastSt; - - public MockInterpreterOutputStream(Properties property) { - super(property); - } - - @Override - public void open() { - //new RuntimeException().printStackTrace(); - } - - @Override - public void close() { - } - - public String getLastStatement() { - return lastSt; - } - - @Override - public InterpreterResult interpret(String st, InterpreterContext context) { - String[] ret = st.split(":"); - try { - if (ret[1] != null) { - context.out.write(ret[1]); - } - } catch (IOException e) { - throw new InterpreterException(e); - } - return new InterpreterResult(InterpreterResult.Code.valueOf(ret[0]), (ret.length > 2) ? - ret[2] : ""); - } - - @Override - public void cancel(InterpreterContext context) { - - } - - @Override - public FormType getFormType() { - return FormType.NATIVE; - } - - @Override - public int getProgress(InterpreterContext context) { - return 0; - } - - @Override - public List<InterpreterCompletion> completion(String buf, int cursor, - InterpreterContext interpreterContext) { - return null; - } - - @Override - public Scheduler getScheduler() { - return SchedulerFactory.singleton().createOrGetFIFOScheduler("interpreter_" + this.hashCode()); - } -} http://git-wip-us.apache.org/repos/asf/zeppelin/blob/8d4902e7/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterResourcePool.java ---------------------------------------------------------------------- diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterResourcePool.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterResourcePool.java deleted file mode 100644 index c4ff6ab..0000000 --- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterResourcePool.java +++ /dev/null @@ -1,128 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.zeppelin.interpreter.remote.mock; - -import java.util.List; -import java.util.Properties; -import java.util.concurrent.atomic.AtomicInteger; - -import com.google.gson.Gson; -import org.apache.zeppelin.display.AngularObjectRegistry; -import org.apache.zeppelin.display.AngularObjectWatcher; -import org.apache.zeppelin.interpreter.Interpreter; -import org.apache.zeppelin.interpreter.InterpreterContext; -import org.apache.zeppelin.interpreter.InterpreterPropertyBuilder; -import org.apache.zeppelin.interpreter.InterpreterResult; -import org.apache.zeppelin.interpreter.InterpreterResult.Code; -import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion; -import org.apache.zeppelin.resource.Resource; -import org.apache.zeppelin.resource.ResourcePool; - -public class MockInterpreterResourcePool extends Interpreter { - - AtomicInteger numWatch = new AtomicInteger(0); - - public MockInterpreterResourcePool(Properties property) { - super(property); - } - - @Override - public void open() { - } - - @Override - public void close() { - - } - - @Override - public InterpreterResult interpret(String st, InterpreterContext context) { - String[] stmt = st.split(" "); - String cmd = stmt[0]; - String noteId = null; - String paragraphId = null; - String name = null; - if (stmt.length >= 2) { - String[] npn = stmt[1].split(":"); - if (npn.length >= 3) { - noteId = npn[0]; - paragraphId = npn[1]; - name = npn[2]; - } else { - name = stmt[1]; - } - } - String value = null; - if (stmt.length >= 3) { - value = stmt[2]; - } - - ResourcePool resourcePool = context.getResourcePool(); - Object ret = null; - if (cmd.equals("put")) { - resourcePool.put(noteId, paragraphId, name, value); - } else if (cmd.equalsIgnoreCase("get")) { - Resource resource = resourcePool.get(noteId, paragraphId, name); - if (resource != null) { - ret = resourcePool.get(noteId, paragraphId, name).get(); - } else { - ret = ""; - } - } else if (cmd.equals("remove")) { - ret = resourcePool.remove(noteId, paragraphId, name); - } else if (cmd.equals("getAll")) { - ret = resourcePool.getAll(); - } else if (cmd.equals("invoke")) { - Resource resource = resourcePool.get(noteId, paragraphId, name); - if (stmt.length >=4) { - Resource res = resource.invokeMethod(value, null, null, stmt[3]); - ret = res.get(); - } else { - ret = resource.invokeMethod(value, null, null); - } - } - - try { - Thread.sleep(500); // wait for watcher executed - } catch (InterruptedException e) { - } - - Gson gson = new Gson(); - return new InterpreterResult(Code.SUCCESS, gson.toJson(ret)); - } - - @Override - public void cancel(InterpreterContext context) { - } - - @Override - public FormType getFormType() { - return FormType.NATIVE; - } - - @Override - public int getProgress(InterpreterContext context) { - return 0; - } - - @Override - public List<InterpreterCompletion> completion(String buf, int cursor, - InterpreterContext interpreterContext) { - return null; - } -} http://git-wip-us.apache.org/repos/asf/zeppelin/blob/8d4902e7/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/NoteInterpreterLoaderTest.java ---------------------------------------------------------------------- diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/NoteInterpreterLoaderTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/NoteInterpreterLoaderTest.java deleted file mode 100644 index 5632513..0000000 --- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/NoteInterpreterLoaderTest.java +++ /dev/null @@ -1,243 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.zeppelin.notebook; - -import java.io.File; -import java.io.IOException; -import java.util.ArrayList; -import java.util.HashMap; - -import com.google.common.collect.Lists; -import com.google.common.collect.Maps; -import org.apache.zeppelin.conf.ZeppelinConfiguration; -import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars; -import org.apache.zeppelin.dep.Dependency; -import org.apache.zeppelin.dep.DependencyResolver; -import org.apache.zeppelin.interpreter.Interpreter; -import org.apache.zeppelin.interpreter.InterpreterFactory; -import org.apache.zeppelin.interpreter.InterpreterInfo; -import org.apache.zeppelin.interpreter.InterpreterOption; -import org.apache.zeppelin.interpreter.DefaultInterpreterProperty; -import org.apache.zeppelin.interpreter.InterpreterProperty; -import org.apache.zeppelin.interpreter.InterpreterSettingManager; -import org.apache.zeppelin.interpreter.LazyOpenInterpreter; -import org.apache.zeppelin.interpreter.mock.MockInterpreter1; -import org.apache.zeppelin.interpreter.mock.MockInterpreter11; -import org.apache.zeppelin.interpreter.mock.MockInterpreter2; -import org.junit.After; -import org.junit.Before; -import org.junit.Test; - -import static org.junit.Assert.*; - -public class NoteInterpreterLoaderTest { - - private File tmpDir; - private ZeppelinConfiguration conf; - private InterpreterFactory factory; - private InterpreterSettingManager interpreterSettingManager; - private DependencyResolver depResolver; - - @Before - public void setUp() throws Exception { - tmpDir = new File(System.getProperty("java.io.tmpdir")+"/ZeppelinLTest_"+System.currentTimeMillis()); - tmpDir.mkdirs(); - new File(tmpDir, "conf").mkdirs(); - - System.setProperty(ConfVars.ZEPPELIN_HOME.getVarName(), tmpDir.getAbsolutePath()); - - conf = ZeppelinConfiguration.create(); - - depResolver = new DependencyResolver(tmpDir.getAbsolutePath() + "/local-repo"); - interpreterSettingManager = new InterpreterSettingManager(conf, depResolver, new InterpreterOption(true)); - factory = new InterpreterFactory(conf, null, null, null, depResolver, false, interpreterSettingManager); - - ArrayList<InterpreterInfo> interpreterInfos = new ArrayList<>(); - interpreterInfos.add(new InterpreterInfo(MockInterpreter1.class.getName(), "mock1", true, Maps.<String, Object>newHashMap())); - interpreterInfos.add(new InterpreterInfo(MockInterpreter11.class.getName(), "mock11", false, Maps.<String, Object>newHashMap())); - ArrayList<InterpreterInfo> interpreterInfos2 = new ArrayList<>(); - interpreterInfos2.add(new InterpreterInfo(MockInterpreter2.class.getName(), "mock2", true, Maps.<String, Object>newHashMap())); - - interpreterSettingManager.add("group1", interpreterInfos, Lists.<Dependency>newArrayList(), new InterpreterOption(), Maps.<String, DefaultInterpreterProperty>newHashMap(), "mock", null); - interpreterSettingManager.add("group2", interpreterInfos2, Lists.<Dependency>newArrayList(), new InterpreterOption(), Maps.<String, DefaultInterpreterProperty>newHashMap(), "mock", null); - - interpreterSettingManager.createNewSetting("group1", "group1", Lists.<Dependency>newArrayList(), new InterpreterOption(), new HashMap<String, InterpreterProperty>()); - interpreterSettingManager.createNewSetting("group2", "group2", Lists.<Dependency>newArrayList(), new InterpreterOption(), new HashMap<String, InterpreterProperty>()); - - - } - - @After - public void tearDown() throws Exception { - delete(tmpDir); - Interpreter.registeredInterpreters.clear(); - } - - @Test - public void testGetInterpreter() throws IOException { - interpreterSettingManager.setInterpreters("user", "note", interpreterSettingManager.getDefaultInterpreterSettingList()); - - // when there're no interpreter selection directive - assertEquals("org.apache.zeppelin.interpreter.mock.MockInterpreter1", factory.getInterpreter("user", "note", null).getClassName()); - assertEquals("org.apache.zeppelin.interpreter.mock.MockInterpreter1", factory.getInterpreter("user", "note", "").getClassName()); - assertEquals("org.apache.zeppelin.interpreter.mock.MockInterpreter1", factory.getInterpreter("user", "note", " ").getClassName()); - - // when group name is omitted - assertEquals("org.apache.zeppelin.interpreter.mock.MockInterpreter11", factory.getInterpreter("user", "note", "mock11").getClassName()); - - // when 'name' is ommitted - assertEquals("org.apache.zeppelin.interpreter.mock.MockInterpreter1", factory.getInterpreter("user", "note", "group1").getClassName()); - assertEquals("org.apache.zeppelin.interpreter.mock.MockInterpreter2", factory.getInterpreter("user", "note", "group2").getClassName()); - - // when nothing is ommitted - assertEquals("org.apache.zeppelin.interpreter.mock.MockInterpreter1", factory.getInterpreter("user", "note", "group1.mock1").getClassName()); - assertEquals("org.apache.zeppelin.interpreter.mock.MockInterpreter11", factory.getInterpreter("user", "note", "group1.mock11").getClassName()); - assertEquals("org.apache.zeppelin.interpreter.mock.MockInterpreter2", factory.getInterpreter("user", "note", "group2.mock2").getClassName()); - - interpreterSettingManager.closeNote("user", "note"); - } - - @Test - public void testNoteSession() throws IOException { - interpreterSettingManager.setInterpreters("user", "noteA", interpreterSettingManager.getDefaultInterpreterSettingList()); - interpreterSettingManager.getInterpreterSettings("noteA").get(0).getOption().setPerNote(InterpreterOption.SCOPED); - - interpreterSettingManager.setInterpreters("user", "noteB", interpreterSettingManager.getDefaultInterpreterSettingList()); - interpreterSettingManager.getInterpreterSettings("noteB").get(0).getOption().setPerNote(InterpreterOption.SCOPED); - - // interpreters are not created before accessing it - assertNull(interpreterSettingManager.getInterpreterSettings("noteA").get(0).getInterpreterGroup("user", "noteA").get("noteA")); - assertNull(interpreterSettingManager.getInterpreterSettings("noteB").get(0).getInterpreterGroup("user", "noteB").get("noteB")); - - factory.getInterpreter("user", "noteA", null).open(); - factory.getInterpreter("user", "noteB", null).open(); - - assertTrue( - factory.getInterpreter("user", "noteA", null).getInterpreterGroup().getId().equals( - factory.getInterpreter("user", "noteB", null).getInterpreterGroup().getId())); - - // interpreters are created after accessing it - assertNotNull(interpreterSettingManager.getInterpreterSettings("noteA").get(0).getInterpreterGroup("user", "noteA").get("noteA")); - assertNotNull(interpreterSettingManager.getInterpreterSettings("noteB").get(0).getInterpreterGroup("user", "noteB").get("noteB")); - - // invalid close - interpreterSettingManager.closeNote("user", "note"); - assertNotNull(interpreterSettingManager.getInterpreterSettings("noteA").get(0).getInterpreterGroup("user", "shared_process").get("noteA")); - assertNotNull(interpreterSettingManager.getInterpreterSettings("noteB").get(0).getInterpreterGroup("user", "shared_process").get("noteB")); - - // when - interpreterSettingManager.closeNote("user", "noteA"); - interpreterSettingManager.closeNote("user", "noteB"); - - // interpreters are destroyed after close - assertNull(interpreterSettingManager.getInterpreterSettings("noteA").get(0).getInterpreterGroup("user", "shared_process").get("noteA")); - assertNull(interpreterSettingManager.getInterpreterSettings("noteB").get(0).getInterpreterGroup("user", "shared_process").get("noteB")); - - } - - @Test - public void testNotePerInterpreterProcess() throws IOException { - interpreterSettingManager.setInterpreters("user", "noteA", interpreterSettingManager.getDefaultInterpreterSettingList()); - interpreterSettingManager.getInterpreterSettings("noteA").get(0).getOption().setPerNote(InterpreterOption.ISOLATED); - - interpreterSettingManager.setInterpreters("user", "noteB", interpreterSettingManager.getDefaultInterpreterSettingList()); - interpreterSettingManager.getInterpreterSettings("noteB").get(0).getOption().setPerNote(InterpreterOption.ISOLATED); - - // interpreters are not created before accessing it - assertNull(interpreterSettingManager.getInterpreterSettings("noteA").get(0).getInterpreterGroup("user", "noteA").get("shared_session")); - assertNull(interpreterSettingManager.getInterpreterSettings("noteB").get(0).getInterpreterGroup("user", "noteB").get("shared_session")); - - factory.getInterpreter("user", "noteA", null).open(); - factory.getInterpreter("user", "noteB", null).open(); - - // per note interpreter process - assertFalse( - factory.getInterpreter("user", "noteA", null).getInterpreterGroup().getId().equals( - factory.getInterpreter("user", "noteB", null).getInterpreterGroup().getId())); - - // interpreters are created after accessing it - assertNotNull(interpreterSettingManager.getInterpreterSettings("noteA").get(0).getInterpreterGroup("user", "noteA").get("shared_session")); - assertNotNull(interpreterSettingManager.getInterpreterSettings("noteB").get(0).getInterpreterGroup("user", "noteB").get("shared_session")); - - // when - interpreterSettingManager.closeNote("user", "noteA"); - interpreterSettingManager.closeNote("user", "noteB"); - - // interpreters are destroyed after close - assertNull(interpreterSettingManager.getInterpreterSettings("noteA").get(0).getInterpreterGroup("user", "noteA").get("shared_session")); - assertNull(interpreterSettingManager.getInterpreterSettings("noteB").get(0).getInterpreterGroup("user", "noteB").get("shared_session")); - } - - @Test - public void testNoteInterpreterCloseForAll() throws IOException { - interpreterSettingManager.setInterpreters("user", "FitstNote", interpreterSettingManager.getDefaultInterpreterSettingList()); - interpreterSettingManager.getInterpreterSettings("FitstNote").get(0).getOption().setPerNote(InterpreterOption.SCOPED); - - interpreterSettingManager.setInterpreters("user", "yourFirstNote", interpreterSettingManager.getDefaultInterpreterSettingList()); - interpreterSettingManager.getInterpreterSettings("yourFirstNote").get(0).getOption().setPerNote(InterpreterOption.ISOLATED); - - // interpreters are not created before accessing it - assertNull(interpreterSettingManager.getInterpreterSettings("FitstNote").get(0).getInterpreterGroup("user", "FitstNote").get("FitstNote")); - assertNull(interpreterSettingManager.getInterpreterSettings("yourFirstNote").get(0).getInterpreterGroup("user", "yourFirstNote").get("yourFirstNote")); - - Interpreter firstNoteIntp = factory.getInterpreter("user", "FitstNote", "group1.mock1"); - Interpreter yourFirstNoteIntp = factory.getInterpreter("user", "yourFirstNote", "group1.mock1"); - - firstNoteIntp.open(); - yourFirstNoteIntp.open(); - - assertTrue(((LazyOpenInterpreter)firstNoteIntp).isOpen()); - assertTrue(((LazyOpenInterpreter)yourFirstNoteIntp).isOpen()); - - interpreterSettingManager.closeNote("user", "FitstNote"); - - assertFalse(((LazyOpenInterpreter)firstNoteIntp).isOpen()); - assertTrue(((LazyOpenInterpreter)yourFirstNoteIntp).isOpen()); - - //reopen - firstNoteIntp.open(); - - assertTrue(((LazyOpenInterpreter)firstNoteIntp).isOpen()); - assertTrue(((LazyOpenInterpreter)yourFirstNoteIntp).isOpen()); - - // invalid check - interpreterSettingManager.closeNote("invalid", "Note"); - - assertTrue(((LazyOpenInterpreter)firstNoteIntp).isOpen()); - assertTrue(((LazyOpenInterpreter)yourFirstNoteIntp).isOpen()); - - // invalid contains value check - interpreterSettingManager.closeNote("u", "Note"); - - assertTrue(((LazyOpenInterpreter)firstNoteIntp).isOpen()); - assertTrue(((LazyOpenInterpreter)yourFirstNoteIntp).isOpen()); - } - - - private void delete(File file){ - if(file.isFile()) file.delete(); - else if(file.isDirectory()){ - File [] files = file.listFiles(); - if(files!=null && files.length>0){ - for(File f : files){ - delete(f); - } - } - file.delete(); - } - } -}
