using System;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;

using NUnit.Framework;

using Python.Runtime;

namespace Python.EmbeddingTest
{
 public class TestInterrupt
 {
 PyObject threading;
 [OneTimeSetUp]
 public void SetUp()
 {
 PythonEngine.Initialize();
 // workaround for assert tlock.locked() warning
 threading = Py.Import("threading");
 }

 [OneTimeTearDown]
 public void Dispose()
 {
 threading.Dispose();
 PythonEngine.Shutdown();
 }

 [Test]
 public void PythonThreadIDStable()
 {
 long pythonThreadID = 0;
 long pythonThreadID2 = 0;
 var asyncCall = Task.Factory.StartNew(() =>
 {
 using (Py.GIL())
 {
 Interlocked.Exchange(ref pythonThreadID, (long)PythonEngine.GetPythonThreadID());
 Interlocked.Exchange(ref pythonThreadID2, (long)PythonEngine.GetPythonThreadID());
 }
 });

 var timeout = Stopwatch.StartNew();

 IntPtr threadState = PythonEngine.BeginAllowThreads();
 while (Interlocked.Read(ref pythonThreadID) == 0 || Interlocked.Read(ref pythonThreadID2) == 0)
 {
 Assert.Less(timeout.Elapsed, TimeSpan.FromSeconds(5), "thread IDs were not assigned in time");
 }
 PythonEngine.EndAllowThreads(threadState);

 Assert.IsTrue(asyncCall.Wait(TimeSpan.FromSeconds(5)), "Async thread has not finished in time");

 Assert.AreEqual(pythonThreadID, pythonThreadID2);
 Assert.NotZero(pythonThreadID);
 }

 [Test]
 public void InterruptTest()
 {
 long pythonThreadID = 0;
 var asyncCall = Task.Factory.StartNew(() =>
 {
 using (Py.GIL())
 {
 Interlocked.Exchange(ref pythonThreadID, (long)PythonEngine.GetPythonThreadID());
 return PythonEngine.RunSimpleString(@"
try:
 import time

 while True:
 time.sleep(0.2)
except KeyboardInterrupt:
 pass");
 }
 });

 var timeout = Stopwatch.StartNew();

 IntPtr threadState = PythonEngine.BeginAllowThreads();
 while (Interlocked.Read(ref pythonThreadID) == 0)
 {
 Assert.Less(timeout.Elapsed, TimeSpan.FromSeconds(5), "thread ID was not assigned in time");
 }
 PythonEngine.EndAllowThreads(threadState);

 int interruptReturnValue = PythonEngine.Interrupt((ulong)Interlocked.Read(ref pythonThreadID));
 Assert.AreEqual(1, interruptReturnValue);

 threadState = PythonEngine.BeginAllowThreads();
 Assert.IsTrue(asyncCall.Wait(TimeSpan.FromSeconds(5)), "Async thread was not interrupted in time");
 PythonEngine.EndAllowThreads(threadState);

 Assert.AreEqual(0, asyncCall.Result);
 }
 }
}